该版本仍在开发中，尚未被视为稳定。对于最新稳定版本，请使用 Spring for Apache Kafka 4.0.0！spring-doc.cadn.net.cn

动态创建容器

在运行时，可以使用多种技术来创建监听器容器。 本节将探讨其中一些技巧。spring-doc.cadn.net.cn

MessageListener 实现

如果你直接实现自己的监听器，你可以直接用容器工厂为该监听器创建一个原始容器：spring-doc.cadn.net.cn

用户监听器
public class MyListener implements MessageListener<String, String> {

    @Override
    public void onMessage(ConsumerRecord<String, String> data) {
        // ...
    }

}

private ConcurrentMessageListenerContainer<String, String> createContainer(
        ConcurrentKafkaListenerContainerFactory<String, String> factory, String topic, String group) {

    ConcurrentMessageListenerContainer<String, String> container = factory.createContainer(topic);
    container.getContainerProperties().setMessageListener(new MyListener());
    container.getContainerProperties().setGroupId(group);
    container.setBeanName(group);
    container.start();
    return container;
}
class MyListener : MessageListener<String, String> {

    override fun onMessage(data: ConsumerRecord<String, String>) {

        // ...
    }

}

private fun createContainer(
    factory: ConcurrentKafkaListenerContainerFactory<String, String>, topic: String, group: String
): ConcurrentMessageListenerContainer<String, String> {
    val container = factory.createContainer(topic)
    container.containerProperties.setMessageListener(MyListener())
    container.containerProperties.setGroupId(group)
    container.beanName = group
    container.start()
    return container
}

原型豆

注释方法容器@KafkaListener可以通过将 Bean 声明为原型来动态创建：spring-doc.cadn.net.cn

原型
public class MyPojo {

    private final String id;

    private final String topic;

    public MyPojo(String id, String topic) {
        this.id = id;
        this.topic = topic;
    }

    public String getId() {
        return this.id;
    }

    public String getTopic() {
        return this.topic;
    }

    @KafkaListener(id = "#{__listener.id}", topics = "#{__listener.topic}")
    public void listen(String in) {
        System.out.println(in);
    }

}

@Bean
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
MyPojo pojo(String id, String topic) {
    return new MyPojo(id, topic);
}

applicationContext.getBean(MyPojo.class, "one", "topic2");
applicationContext.getBean(MyPojo.class, "two", "topic3");
class MyPojo(val id: String, val topic: String) {

    @KafkaListener(id = "#{__listener.id}", topics = ["#{__listener.topic}"])
    fun listen(`in`: String?) {
        println(`in`)
    }

}

@Bean
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
fun pojo(id: String, topic: String): MyPojo {
    return MyPojo(id, topic)
}

applicationContext.getBean(MyPojo::class.java, "one", "topic2")
applicationContext.getBean(MyPojo::class.java, "two", "topic3")
听众必须拥有唯一的ID。 从2.8.9版本开始，KafkaListenerEndpointRegistry有一种新方法unregisterListenerContainer（String id）这样你就可以重复使用身份证。 取消注册容器不会停止（）容器，你得自己动手。