@kafkaListener 的范围

Question

我只是想了解@kafkaListener 的范围是什么，无论是原型还是单例。如果单个主题有多个消费者，它是返回单个实例还是多个实例。就我而言，我有多个客户订阅了单个主题并获取报告。我只是想知道，如果

• 多个客户想同时查询报表。在 我的情况，我在成功消费后关闭容器 消息，但同时如果其他人想要获取 报告，容器应该是打开的。

• 如何将范围更改为与 Id 关联的原型（如果不是） 容器，这样每次都可以生成一个单独的实例。

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

Answer 1

为所有消费线程调用单个侦听器实例。

注解@KafkaListener 不是Prototype 作用域的，这个注解也不可能。

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/

=== 已编辑 ===

让我们采用他们的第三个选项（Delcaring a 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
    }
}

创建@Service

public class KafkaListenerService{


    @Autowired
    private KafkaDelegate kafkaDelegate;

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

}

另一个例子：How to implement a stateful message listener using Spring Kafka?

Answer 2

请参阅 this answer 以获取有关如何使用范围为 @KafkaListener bean 的原型的示例。