@kafkaListener的范围

Question

我只想了解@kafkaListener的范围是什么，原型还是单例。在单个主题的多个消费者的情况下，返回的是单个实例还是多个实例。在我的情况下，我有多个客户订阅单个主题并获得报告。我只是想知道如果

>

多个客户希望同时查询报告。在我的例子中，我在成功使用消息后关闭容器，但同时如果其他人想要获取报告，则容器应该打开。

如何将作用域更改为与容器的id相关联的原型（如果不是），以便每次都可以生成单独的实例。

@KafkaListener(id = "id1", topics = "testTopic" )
 public void listen() {
    // code goes here
}

Answer 1

有关如何使用原型范围的@KafkaListenerbean的示例，请参见此答案。

Answer 2

为所有使用线程调用一个侦听器实例。

注释@KafkaListener不在原型范围内，此注释也不可能。

4.1.10. Thread Safety

When using a concurrent message listener container, a single listener instance is invoked on all consumer threads. Listeners, therefore, need to be thread-safe, and it is preferable to use stateless listeners. If it is not possible to make your listener thread-safe or adding synchronization would significantly reduce the benefit of adding concurrency, you can use one of a few techniques:

    Use n containers with concurrency=1 with a prototype scoped MessageListener bean so that each container gets its own instance (this is not possible when using @KafkaListener).

    Keep the state in ThreadLocal<?> instances.

    Have the singleton listener delegate to a bean that is declared in SimpleThreadScope (or a similar scope).

To facilitate cleaning up thread state (for the second and third items in the preceding list), starting with version 2.2, the listener container publishes a ConsumerStoppedEvent when each thread exits. You can consume these events with an ApplicationListener or @EventListener method to remove ThreadLocal<?> instances or remove() thread-scoped beans from the scope. Note that SimpleThreadScope does not destroy beans that have a destruction interface (such as DisposableBean), so you should destroy() the instance yourself.
    By default, the application context’s event multicaster invokes event listeners on the calling thread. If you change the multicaster to use an async executor, thread cleanup is not effective. 

https://docs.spring.io/spring-kafka/reference/html/

==已编辑===

让我们选择第三个选项（删除SimpleThreadScope并授权给它）

注册SimpleThreadScope。它不会自动拾取。您需要注册如下：

@Bean
public static BeanFactoryPostProcessor beanFactoryPostProcessor() {
    return new BeanFactoryPostProcessor() {
        @Override
        public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {

            beanFactory.registerScope("thread", new SimpleThreadScope());
        }
    };
}

使用scopeName=“thread”创建组件

    @Component
    @Scope(scopeName = "thread", proxyMode = ScopedProxyMode.TARGET_CLASS)
    public class KafkaDelegate{


     public void handleMessageFromKafkaListener(String message){
  
             //Do some stuff here with Message
    }
}

创建@服务

public class KafkaListenerService{


    @Autowired
    private KafkaDelegate kafkaDelegate;

    
    @KafkaListener(id = "id1", topics = "testTopic" )
    public void listen(String message) {
        kafkaDelete.handleMessageFromKafkaListener(message);
    }

}

另一个示例：如何使用SpringKafka实现有状态消息侦听器？