如何使用Spring-Kafka实现消费者线程安全

Question

我正在使用Spring boot2.1.7。RELEASE和spring-kafka 2.2.7。RELEASE。我正在使用@KafkaListener注释来创建一个消费者，我正在使用消费者的所有默认设置。

这是我的消费者配置：

@Configuration
@EnableKafka
public class KafkaConsumerCommonConfig implements KafkaListenerConfigurer {

      @Bean
      public <K,V> ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(primaryConsumerFactory());
        factory.getContainerProperties().setMissingTopicsFatal(false);
        return factory;
      }

      @Bean
      public DefaultKafkaConsumerFactory<Object, Object> primaryConsumerFactory() {
        return new DefaultKafkaConsumerFactory<>(sapphireKafkaConsumerConfig.getConfigs());
      }

}

由于某些原因，我在同一个应用程序中有多个使用者，如下所示。

@KafkaListener(topics = "TEST_TOPIC1")
public void consumer1(){

}

@KafkaListener(topics = "TEST_TOPIC2")
public void consumer2(){

}

@KafkaListener(topics = "TEST_TOPIC3")
public void consumer3(){

}

尽管如此，根据关于“消费者线程安全”的合流文件

一个线程中不能有多个属于同一组的使用者，也不能有多个线程安全地使用同一使用者。每个线程一个使用者是规则。要在一个应用程序中运行同一组中的多个使用者，需要在各自的线程中运行每个使用者。将消费者逻辑封装在自己的对象中，然后使用Java的ExecutorService启动多个线程，每个线程都有自己的消费者，这是很有用的。

现在我的问题是，当使用上面的代码使用spring kafka来解决这个场景时，我应该做些什么额外的事情吗？还是不重要，因为我没有指定组是随机生成的？请提出建议。

Answer 1

您的所有侦听器都针对不同的主题进行了配置。所以，他们的团队完全没有关系。当同一主题有多个消费者时，团队就会出现。