如何在应用程序属性中拥有多个kafka消费群体

Question

我正在使用spring kafka并希望在应用程序yaml中提到多个消费者群组，我可以在我的kafkaListener类中使用这些群组，我正在听多个主题。

application.yml文件中的kafka属性现在如下所示

kafka:
    properties:
      topics:
        topic1: topic1
        topic2: topic2
    bootstrap-servers: server1,server2
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      retries: 4
    consumer:
      group-id: mygroupid
      auto-offset-reset: latest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

有没有办法在上面的使用者块中有多个群组。

 @KafkaListener(topics = "topic1",  groupId = "cd1")
  public void consumeMessage(String message) throws Exception {
    // some code goes here
  }

Answer 1

引导仅从yml自动配置一个DefaultKafkaConsumerFactory和DefaultKafkaConsumerFactory，因此所有属性在所有使用者之间共享。这就是为什么我们添加了groupid（如果未提供groupid，则使用id)；这是消费者之间变化最常见的属性。

当然，您可以使用属性占位符，因此groupid=“${group.one}”将使用YML中的属性group.one。

要更改更基本的东西，如序列化器/反序列化器，如果您使用的版本早于2.2.4，则需要创建多个工厂和容器工厂。

但是，从版本2.2.4开始，您现在可以在kafkalistener注释中设置任何任意的kafka使用者属性...

/**
 * Kafka consumer properties; they will supersede any properties with the same name
 * defined in the consumer factory (if the consumer factory supports property overrides).
 * <h3>Supported Syntax</h3>
 * <p>The supported syntax for key-value pairs is the same as the
 * syntax defined for entries in a Java
 * {@linkplain java.util.Properties#load(java.io.Reader) properties file}:
 * <ul>
 * <li>{@code key=value}</li>
 * <li>{@code key:value}</li>
 * <li>{@code key value}</li>
 * </ul>
 * {@code group.id} and {@code client.id} are ignored.
 * @return the properties.
 * @since 2.2.4
 * @see org.apache.kafka.clients.consumer.ConsumerConfig
 * @see #groupId()
 * @see #clientIdPrefix()
 */
String[] properties() default {};

注意，这些属性是本机的kafka虚线格式(auto.offset.reset)，而不是boot连字符或camel大小写属性。

下面是文档中的一个示例：

@KafkaListener(topics = "myTopic", groupId="group", properties= {
    "max.poll.interval.ms:60000",
    ConsumerConfig.MAX_POLL_RECORDS_CONFIG + "=100"
})