spring-kafka之请求响应模式-Toy模板网

这篇具有很好参考价值的文章主要介绍了spring-kafka之请求响应模式。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

kafka是一款性能强劲的分布式流式处理软件，被广泛用于大数据应用场景。所以很多小伙伴对kafka肯定不会陌生，但是kafka的请求响应模式估计使用的却不一定很多。首先简单唠叨下什么是请求响应模式，这个类似于http请求一样发出请求能够在一个请求中返回结果，所以这种场景跟小伙伴大部分使用kafka的场景肯定不大一样，但是这种模式却可以简化下述场景的使用：

场景：数据删除校验

随着微服务化的发展，很多数据不再像最初的单库模式一样都放在一个数据库实例里，并不是所有的服务都在一个服务里。所以删除某些数据时无法再像单体模式一样进行本地校验，也有有人会说可以通过微服务调用的方式来进行校验，确实微服务确实可以实现删除校验的功能，但是当服务越来越多并且逐渐增长的时候就会出现问题，导致被依赖方需要频繁的调整代码，因为依赖这个基础服务的其他服务一直在变化。所以这个是否使用mq进行数据校验的解耦就成为一种很好的替代方案。相信大部分人使用mq实现该功能的方案就是创建两个topic：请求topic以及响应topic，基础服务删除数据前向请求topic发送数据，服务依赖方收到对应的删除校验请求后判断该服务是否有数据依赖删除的基础数据，如果则向响应topic发送数据。

我们可以看到上面的交互用到了两个topic，并且鉴于上述响应的异步性，删除校验端需要启动异步处理等待响应的返回，同时需要启动超时检测机制（不能一直等待），这种双topic确实可以解决这种删除校验的逻辑，但是实现比较繁琐，今天咱们就来探索下kafka另外一个处理方式：请求响应模式，看下这种模式如何简化处理流程的。

kafka实现请求响应在spring框架下很容易实现，ReplyingKafkaTemplate这个类就可以实现该功能，废话不多说，直接给出实例代码：


    @Autowired
    private ReplyingKafkaTemplate<String, String, String> replyingKafkaTemplate;
    
    @Test
    public void sendReplyMsg() throws JsonProcessingException, InterruptedException {
        doSendReplyMsg("si", "li");
        doSendReplyMsg("san", "zhang");
        TimeUnit.SECONDS.sleep(50);
    }
    
    public void doSendReplyMsg(String key, String msg) throws JsonProcessingException, InterruptedException {
        ProducerRecord<String, String> request = new ProducerRecord<>("reply", key, msg);
        // 发送数据并设置3秒的响应超时时间，这个时间可以根据需要自己设定
        RequestReplyFuture<String, String, String> future = replyingKafkaTemplate.sendAndReceive(request,
                Duration.of(3, ChronoUnit.SECONDS));
        future.addCallback(new ListenableFutureCallback<ConsumerRecord<String, String>>() {
            @Override
            public void onFailure(Throwable ex) {
            
            }
            
            @Override
            public void onSuccess(ConsumerRecord<String, String> result) {
                log.info("发送数据：key:{},value:{}收到响应:{}", key, msg, result.value());
            }
        });
    }

@Slf4j
@EnableKafka
@Configuration
public class KafkaConfig {
    
    /**
    ** 定义ReplyingKafkaTemplate.
    **/
    @Bean
    public ReplyingKafkaTemplate<String, String, String> replyKafkaTemplate(ProducerFactory<String, String> pf,
            ConcurrentMessageListenerContainer<String, String> repliesContainer) {
        ReplyingKafkaTemplate<String, String, String> replyTemplate = new ReplyingKafkaTemplate<>(pf, repliesContainer);
        replyTemplate.setDefaultReplyTimeout(Duration.ofSeconds(3));
        replyTemplate.setSharedReplyTopic(true);
        return replyTemplate;
    }
    
    @Bean
    public ConcurrentMessageListenerContainer<String, String> repliesContainer(ConsumerFactory<String, String> cf) {
        ContainerProperties containerProperties = new ContainerProperties("repliesTopic");
        return new ConcurrentMessageListenerContainer<>(cf, containerProperties);
    }
    
    @Bean
    public KafkaTemplate<String, String> kafkaTemplate(ProducerFactory<String, String> pf) {
        return new KafkaTemplate<>(pf);
    }
}

上述代码为kafka消息生产者相关，下面是kafka消费者代码：

@Component
public class KafkaListeners {
    
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    
    @KafkaListener(topics = "reply", groupId = "111")
    public void listen(ConsumerRecord<String, String> record, @Headers MessageHeaders headers) {
        log.info("listen收到数据");
        Object obj = headers.get(KafkaHeaders.REPLY_TOPIC);
        if (null != obj) {
            String replyTopic = new String((byte[]) obj);
            RecordHeaders headers1 = new RecordHeaders();
            // 返回数据里面需要在header中增加kafka_correlationId
            headers1.add(KafkaHeaders.CORRELATION_ID, record.headers().lastHeader(KafkaHeaders.CORRELATION_ID).value());
            ProducerRecord<String, String> record1 = new ProducerRecord<>(replyTopic, record.partition(), record.key(),
                    record.value() + record.key(), headers1);
            kafkaTemplate.send(record1);
        }
    }
}

注意上述备注部分，kafka消费端需要在kafka的ProducerRecord header中增加kafka_correlationId，而且该字段需要跟发送方发送的kafka_correlationId值保持一致，这也是生产端进行消息匹配的值。

但需要注意的是及时采用的是kafka的topic模式，多个消费者可能都会响应，但是生产端在收到一个数据后就不再接收后续消费者发送的响应，ReplyingKafkaTemplate的源码可以参考：ReplyingKafkaTemplate源码分析_PolarisHuster的博客-CSDN博客文章来源地址https://www.toymoban.com/news/detail-508615.html

到了这里，关于spring-kafka之请求响应模式的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！