Spring Cloud Stream功能生产者/消费者

Question

我的应用程序由一个带有POST方法的REST控制器组成，用于提交我必须使用生产者发送到主题的数据。

这是控制器

@RestController
@RequestMapping("/producer/v1")
public class ApiController {
    @Autowired
    Producer producer;

    @ApiOperation(value = "Invia un messaggio al topic", response = String.class)
    @ApiResponses(value = {
            @ApiResponse(code = 200, message = "Success"),
            @ApiResponse(code = 400, message = "Bad Request"),
            @ApiResponse(code = 500, message = "Failure")})
    @PostMapping(path = "/sendMessage", consumes = "application/json", produces = "application/json")
    public void sendMessage(@RequestBody() @Valid MyMessage myMessage) {
        producer.produce(myMessage.getId(), myMessage);
    }
}

使用Spring-Cloud-Stream版本

@Component
public class Producer {
    private final ProducerChannelInterface producerChannelInterface;

    public Producer(ProducerChannelInterface producerChannelInterface) {
        this.producerChannelInterface = producerChannelInterface;
    }
    public void produce(int messageId, Object message) {
        MessageChannel messageChannel = producerChannelInterface.kafkaOutChannel();
        messageChannel.send(
                MessageBuilder
                        .withPayload(message)
                        .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                        .setHeader(PARTITION_KEY, messageId)
                        .build()
        );
    }
}

从3.1版开始，@Output和@EnableBinding注释被弃用，所以我尝试切换到新的方式来设置生产者，我就是这样工作的

@Component
public class Producer {
    @Bean
    public Supplier<Message<Object>> produce() {
        return () -> {
            int messageId = ...;
            Object message = new MyMessage();
            return MessageBuilder
                .withPayload(message)
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                .setHeader(PARTITION_KEY, messageId)
                .build();
        };
    }
}

最后在应用程序中。yaml我有这个

spring:
  cloud:
    stream:
      bindings:
        #Channel name
        produce-out-0:
          destination: spring-cloud-topic
          contentType: application/json
          producer:
            partitionKeyExpression: headers['partitionKey']
            partitionCount: 1
            errorChannelEnabled: true

...
      kafka:
        bindings:
          produce-out-0:
            producer:
              configuration:
                retries: 10
                max.in.flight.requests.per.connection: 1
                request.timeout.ms: 20000

现在的问题是，当我启动应用程序时，方法产生（）被无限调用（我在主题中看到消息）。然后使用供应商似乎我被迫在供应商内部定义消息数据。我需要从Rest控制器传递数据。

你能帮忙吗？谢谢

Answer 1

普通的SCS3供应商有几个“问题”，但您可以解决这些问题。主要问题是投票。默认情况下，SCS将每隔几秒钟轮询一次消息。当然，您可以使用DefaultPollerProperties来配置它。但是，当它轮询消息时，您必须提供一个。如果您可以设置适当的轮询超时，那么您可以使用队列来提供消息，并且在供应商中，您可以对该队列使用轮询。但这不是正确的解决方案，因为在rest中，您无法计算消息之间的延迟，您希望立即发布消息，当然，不要等待rest调用之间的最长时间。。。更好的方法是使用反应流。因此，当消息通过rest接口到达时，您可以发送消息，然后消息将立即发布。

您可以使用StreamBridge将消息推送到任何流中。这会将消息推送到流中，但流必须存在（非反应流也存在同样的问题）。你可以把它注入到你的服务中，并通过网桥发送信息。请注意，它接受一个对象参数，因此必须为给定流提供正确的类型。

@Autowired
private lateinit var bridge: StreamBridge

然后你就可以打电话了

bridge.send("produce-out-0", messageObject)

反应流是更好的方法。

您可以使用前面的空Flux方法。这不会发出任何消息。所以您只能使用StreamBridge发布消息。

@Bean
public Supplier<Flux<Message<MyType>>> produce() {
    return () -> {
      Flux.empty<>()
    };
}

更好的方法是创建一个接收器，并通过以下方式发送消息：

Sinks.Many<Object> processor = Sinks.many().unicast().onBackpressureBuffer<Object>()

然后你可以用它作为供应商的来源

@Bean
public Supplier<Flux<Message<Object>>> produce() {
    return () -> {
      processor.asFlux()
    };
}

并在其中插入一条信息：

 processor.emitNext(messageObject, Sinks.EmitFailureHandler.FAIL_FAST)