Spring Kafka:ApplicationContext 中不同对象的多个侦听器

本文介绍了Spring Kafka:ApplicationContext 中不同对象的多个侦听器的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！

问题描述

限时送ChatGPT账号..

我能否向社区询问收听多个主题的最佳方式是什么，每个主题包含不同类别的消息?

Can I please check with the community what is the best way to listen to multiple topics, with each topic containing a message of a different class?

过去几天我一直在玩 Spring Kafka.到目前为止我的思考过程:

I've been playing around with Spring Kafka for the past couple of days. My thought process so far:

因为在初始化 KafkaListenerContainerFactory 时需要将反序列化器传递给 DefaultKafkaConsumerFactory.这似乎表明，如果我需要多个容器，每个容器反序列化不同类型的消息，我将无法使用 @EnableKafka 和 @KafkaListener 注释.

Because you need to pass your deserializer into DefaultKafkaConsumerFactory when initializing a KafkaListenerContainerFactory. This seems to indicate that if I need multiple containers each deserializing a message of a different type, I will not be able to use the @EnableKafka and @KafkaListener annotations.

这让我认为唯一的方法是实例化多个 KafkaMessageListenerContainers.

This leads me to think that the only way to do so would be to instantiate multiple KafkaMessageListenerContainers.

鉴于 KafkaMessageListenerContainers 是单线程的并且我需要同时收听多个主题，我真的应该使用多个 ConcurrentKafkaMessageListenerContainers.

And given that KafkaMessageListenerContainers is single threaded and I need to listen to multiple topics at the same time, I really should be using multiple ConcurrentKafkaMessageListenerContainers.

我会在正确的轨道上吗?有没有更好的方法来做到这一点?

Would I be on the right track here? Is there a better way to do this?

谢谢！

推荐答案

这是一个非常简单的例子.

Here is a very simple example.

// -----------------------------------------------
// Sender
// -----------------------------------------------

@Configuration
public class SenderConfig {
    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        ......
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        return props;
    }

    @Bean
    public ProducerFactory<String, Class1> producerFactory1() {
        return new DefaultKafkaProducerFactory<String, Class1>(producerConfigs());
    }

    @Bean
    public KafkaTemplate<String, Class1> kafkaTemplate1() {
        return new KafkaTemplate<>(producerFactory1());
    }

    @Bean
    public Sender1 sender1() {
        return new Sender1();
    }

    //-------- send the second class --------

    @Bean
    public ProducerFactory<String, Class2> producerFactory2() {
        return new DefaultKafkaProducerFactory<String, Class2>(producerConfigs());
    }

    @Bean
    public KafkaTemplate<String, Class2> kafkaTemplate2() {
        return new KafkaTemplate<>(producerFactory2());
    }

    @Bean
    public Sender2 sender2() {
        return new Sender2();
    }
}

public class Sender1 {
    @Autowired
    private KafkaTemplate<String, Class1> kafkaTemplate1;

    public void send(String topic, Class1 c1) {
        kafkaTemplate1.send(topic, c1);
   }
}

public class Sender2 {
    @Autowired
    private KafkaTemplate<String, Class2> kafkaTemplate2;

    public void send(String topic, Class2 c2) {
        kafkaTemplate2.send(topic, c2);
    }
}

// -----------------------------------------------
// Receiver
// -----------------------------------------------

@Configuration
@EnableKafka
public class ReceiverConfig {

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        ......
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        return props;
    }

    @Bean
    public ConsumerFactory<String, Class1> consumerFactory1() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs(), new StringDeserializer(),
            new JsonDeserializer<>(Class1.class));
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Class1> kafkaListenerContainerFactory1() {
        ConcurrentKafkaListenerContainerFactory<String, Class1> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory1());
        return factory;
    }

    @Bean
    public Receiver1 receiver1() {
        return new Receiver1();
    }

    //-------- add the second listener

    @Bean
    public ConsumerFactory<String, Class2> consumerFactory2() {
    return new DefaultKafkaConsumerFactory<>(consumerConfigs(), new StringDeserializer(),
            new JsonDeserializer<>(Class2.class));
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Class2> kafkaListenerContainerFactory2() {
        ConcurrentKafkaListenerContainerFactory<String, Class2> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory2());
        return factory;
    }

    @Bean
    public Receiver2 receiver2() {
        return new Receiver2();
    }
}

public class Receiver1 {
    @KafkaListener(id="listener1", topics = "topic1", containerFactory = "kafkaListenerContainerFactory1")
    public void receive(Class1 c1) {
        LOGGER.info("Received c1");
    }
}

public class Receiver2 {
    @KafkaListener(id="listener2", topics = "topic2", containerFactory = "kafkaListenerContainerFactory2")
    public void receive(Class2 c2) {
        LOGGER.info("Received c2");
    }
}

这篇关于Spring Kafka:ApplicationContext 中不同对象的多个侦听器的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！