Spring Boot kafkaTemplate 消费者消息加载和处理消息

Question

在我的应用程序中，我使用Spring Boot kafkaTemplate 来使用消息。我是使用 Spring Boot 的 kafka 新手。我添加了一个消费者如下 -

 @KafkaListener(topics = "#{'${app.kafka.consumer.topic}'.split(',')}")
 public void receivedMessage(ConsumerRecord<String, String> cr, @Payload String message){
    log.info("Message received from topic {} ", cr.topic());
    //TODO
}

在topic 上，我们每秒将收到近 20 万条消息。我收到的message 将发送到另一个处理方法，该方法根据特定条件过滤消息​​，然后将过滤后的message 发布到另一个topic。

我的问题是，上述@KafkaListener 方法是否会处理此负载，或者我是否需要进行任何特殊处理，例如threading 或concurrency。

Answer 1

这完全取决于您的工作负载、CPU 中的内核数量等。一般来说，增加并发性将提供更多的吞吐量（只要您在主题上至少有那么多分区）。

但是，您的下游代码必须是完全线程安全的。

即便如此，您的代码中也可能存在其他瓶颈（数据库等）。

如果您所做的只是计算、转换和发布到另一个主题，而不进行任何其他 I/O，那么增加并发肯定会有所帮助。

唯一真正的解决方案是实验，如果您没有获得所需的吞吐量，请分析您的应用程序。

如果不实际操作，您无法学习这些技能。