Spring启动Kafka请求-回复场景

Question

我正在实现请求/应答场景的POC，以便使用Kafka移动基于事件的微服务堆栈。

Spring有两种选择。我不知道哪一个更好用<代码>回复Kafka模板或云流

第一个是回复KafkaTemplate，它可以很容易地配置为有专门的频道来回复每个实例的主题<代码>记录。标题（）。添加（新记录头（KafkaHeaders.REPLY_主题，provider.getReplyChannelName（）。getBytes（）） 消费者不需要知道回复主题的名称，只需收听主题并返回给定的回复主题即可。

@KafkaListener(topics = "${kafka.topic.concat-request}")
@SendTo
public ConcatReply listen(ConcatModel request) {
       .....
}

第二种选择是结合使用StreamListener、spring集成和IntegrationFlows。应配置网关并筛选回复主题。

@MessagingGateway
public interface StreamGateway {
    @Gateway(requestChannel = START, replyChannel = FILTER, replyTimeout = 5000, requestTimeout = 2000)
    String process(String payload);
}
@Bean
public IntegrationFlow headerEnricherFlow() { 
    return IntegrationFlows.from(START)
            .enrichHeaders(HeaderEnricherSpec::headerChannelsToString)
            .enrichHeaders(headerEnricherSpec -> headerEnricherSpec.header(Channels.INSTANCE_ID ,instanceUUID)) 
            .channel(Channels.REQUEST)
            .get();
}
@Bean
public IntegrationFlow replyFiltererFlow() {
    return IntegrationFlows.from(GatewayChannels.REPLY)
            .filter(Message.class, message -> Channels.INSTANCE_ID.equals(message.getHeaders().get("instanceId")) )
            .channel(FILTER)
            .get();
}

建筑答复

@StreamListener(Channels.REQUEST)
@SendTo(Channels.REPLY)
public Message<?> process(Message<String> request) { 

必须指定回复渠道。所以收到的回复主题会根据instanceID进行过滤，这是一种解决方案（可能会使网络膨胀）。另一方面，DLQ场景是通过添加

              consumer:
                enableDlq: true

在与RabbitMQ和其他功能的互操作性方面，使用SpringCloudStreams看起来很有希望，但目前还没有正式支持请求-应答场景。问题仍然悬而未决，未被驳回。(https://github.com/spring-cloud/spring-cloud-stream/issues/1800)

欢迎提出任何建议。

Answer 1

Spring Cloud Stream不是为请求/回复而设计的；它可以完成，但不简单，你必须编写代码。

使用@KafkaListener框架可以为您处理所有事情。

如果您希望它也能与RabbitMQ一起使用，还可以使用@RabbitListener对其进行注释。