具有相同组id的多个kafka消费者

Question

我对Kafka是陌生的。我用spring boot创建了一个kafka消费者(spring-kafka dependency)。在我的应用程序中，我使用了consumerFactory和producerfactory beans进行配置。所以在我的应用html" target="_blank">程序中，我创建了如下的kafka消费者。

@RetryableTopic(
        attempts = "3",
        backoff = @Backoff(delay = 1000, multiplier = 2.0),
        autoCreateTopics = "false")
  @KafkaListener(topics = "myTopic", groupId = "myGroupId")
  public void consume(@Payload(required = false) String message) {

       processMessage(message);
  
}

我的配置如下

@Bean
  public ConsumerFactory<String, Object> consumerFactory() {
    Map<String, Object> config = new HashMap<>();
    config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, env.getProperty("kafka.consumer.bootstrap.servers"));
    config.put(ConsumerConfig.GROUP_ID_CONFIG, env.getProperty("kafka.consumer.group"));
    config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

    return new DefaultKafkaConsumerFactory<>(config);
  }

  @Bean
  public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, Object> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.getContainerProperties().setCommitLogLevel(LogIfLevelEnabled.Level.DEBUG);
    factory.getContainerProperties().setMissingTopicsFatal(false);
    return factory;
  }

  @Bean
  public ProducerFactory<String, String> producerFactory() {
    Map<String, Object> config = new HashMap<>();
    config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, env.getProperty("kafka.consumer.bootstrap.servers"));
    config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    return new DefaultKafkaProducerFactory<>(config);
  }

  @Bean
  public KafkaTemplate<String, String> kafkaTemplate() {
    return new KafkaTemplate<>(producerFactory());
  }

所以我想并行消费，因为我可能会收到更多的消息。关于使用并行主题，我发现我需要为一个主题创建多个分区，并且需要为每个分区创建一个使用者。假设我的主题有10个分区，那么我可以让同一消费者组中的10个消费者分别读取一个分区。我理解这种行为。但我担心的是如何在应用程序中创建几个消费者。

我是否必须使用具有相同功能的@KafkaListener编写多个kafka消费者？在这种情况下，如果我需要X个相同的消费者，我需要在下面写X次方法。

@RetryableTopic(
            attempts = "3",
            backoff = @Backoff(delay = 1000, multiplier = 2.0),
            autoCreateTopics = "false")
      @KafkaListener(topics = "myTopic", groupId = "myGroupId")
      public void consume(@Payload(required = false) String message) {
    
           processMessage(message);
      
    }

我需要哪些选项或配置来实现与多个使用者的并行消费？

先谢谢你。

Answer 1

< code>@KafkaListener有这个选项:

/**
 * Override the container factory's {@code concurrency} setting for this listener. May
 * be a property placeholder or SpEL expression that evaluates to a {@link Number}, in
 * which case {@link Number#intValue()} is used to obtain the value.
 * <p>SpEL {@code #{...}} and property place holders {@code ${...}} are supported.
 * @return the concurrency.
 * @since 2.2
 */
String concurrency() default "";

查看更多文档：https://docs.spring.io/spring-kafka/reference/html/#kafka-listener-annotation