Spring Kafka 支持多消费者的需求

Question

正如人们所期望的那样，希望让不同的消费者以不同的方式反序列化 Kafka 中的主题是很常见的。 spring boot 自动配置存在一个已知问题。似乎一旦定义了其他工厂，Spring Kafka 或 autoconfig 就会抱怨无法再找到合适的消费者工厂。有人指出，一种解决方案是在配置中包含一个类型为 (Object, Object) 的 ConsumerFactory。但是没有人展示它的源代码或澄清是否需要以任何特定方式命名。或者，如果只是将此使用者添加到配置中，则无需关闭自动配置。所有这些都还不清楚。

如果您对此问题不熟悉，请阅读https://github.com/spring-projects/spring-boot/issues/19221

在刚刚声明好的地方，定义 ConsumerFactory 并将其添加到配置中的某个位置。请问有人能说得详细一点吗？

1. 准确展示如何定义 ConsumerFactory，以便 Spring boot 自动配置不会报错。
2. 说明是否需要关闭自动配置？
3. 说明消费者工厂是否需要以任何特殊方式命名。

Answer 1

最简单的解决方案是坚持使用 Boot 的自动配置并覆盖 @KafkaListener 本身的反序列化器...

@SpringBootApplication
public class So63108344Application {

    public static void main(String[] args) {
        SpringApplication.run(So63108344Application.class, args);
    }

    @KafkaListener(id = "so63108344-1", topics = "so63108344-1")
    public void listen1(String in) {
        System.out.println(in);
    }

    @KafkaListener(id = "so63108344-2", topics = "so63108344-2", properties =
            ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG +
                "=org.apache.kafka.common.serialization.ByteArrayDeserializer")
    public void listen2(byte[] in) {
        System.out.println(in);
    }

    @Bean
    public NewTopic topic1() {
        return TopicBuilder.name("so63108344-1").partitions(1).replicas(1).build();
    }

    @Bean
    public NewTopic topic2() {
        return TopicBuilder.name("so63108344-2").partitions(1).replicas(1).build();
    }

}

对于更高级的容器定制（或者如果您不想污染@KafkaListener，您可以使用ContainerCustomizer...

@Component
class Customizer {
    
    public Customizer(ConcurrentKafkaListenerContainerFactory<?, ?> factory) {
        factory.setContainerCustomizer(container -> {
            if (container.getGroupId().equals("so63108344-2")) {
                container.getContainerProperties().setAckMode(AckMode.MANUAL_IMMEDIATE);
                container.getContainerProperties().getKafkaConsumerProperties()
                    .setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");
            }
        });
    }
    
}