@KafkaListener作用域是singleton还是prototype？多消费者场景如何配置？

一、@KafkaListener默认作用域说明 #

• 默认情况下，包含@KafkaListener注解的Bean是Singleton单实例，对应的监听容器（MessageListenerContainer）也是单实例，同一个topic下的多个消费者默认是在单实例容器中以多线程方式运行，不会生成多个独立的Bean实例。
• 未手动修改作用域配置时，即便有多个消费者消费同一个topic，返回的都是同一个监听Bean实例。

二、多用户查询场景容器可用性保障方案 #

你当前消费成功就关闭容器的逻辑会导致后续用户请求无可用容器，可选择以下两种适配方案：

• 推荐方案：调整生命周期逻辑，不要在消费完成后关闭容器，让监听容器长期后台运行，每次消费对应单个用户的报表查询请求，消费完成后仅返回报表结果给对应用户，容器保持运行接收后续请求，天然支持多用户并发场景。
• 兼容现有逻辑方案：新增容器状态检测和启停控制逻辑，每次用户发起报表查询请求时，先通过KafkaListenerEndpointRegistry获取对应容器的运行状态，如果容器已停止就重新启动容器；如果需要多用户完全隔离，也可以每次请求都新建临时监听容器，消费完成后销毁当前临时容器，不会影响其他用户的容器实例。

三、将@KafkaListener修改为prototype作用域的实现方式 #

如果需要每次生成独立的监听实例，可通过以下两种方式实现：

方式1：修改监听Bean作用域配合动态ID #

给包含@KafkaListener的类添加@Scope注解声明为prototype作用域，同时用SpEL表达式给@KafkaListener生成动态唯一ID，保证每次初始化Bean时都生成新的独立监听容器：

import org.springframework.context.annotation.Scope;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.context.annotation.ScopedProxyMode;

// 声明为prototype作用域，每次获取Bean都生成新实例
@Component
@Scope(value = "prototype", proxyMode = ScopedProxyMode.TARGET_CLASS)
public class ReportListener {

    // 动态生成唯一的listener ID，每个实例对应独立的容器
    @KafkaListener(id = "#{T(java.util.UUID).randomUUID().toString()}", topics = "testTopic" )
    public void listen() {
        // 你的报表消费逻辑
    }
}

方式2：手动创建独立监听容器（多用户隔离场景更推荐） #

直接通过KafkaListenerContainerFactory手动创建和管理监听容器，每次调用都会生成完全独立的容器实例，不需要依赖注解驱动，生命周期完全可控：

import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class ReportContainerService {

    @Autowired
    private KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> containerFactory;

    // 每次调用生成新的独立监听容器
    public ConcurrentMessageListenerContainer<String, String> createIndependentContainer(String customerId) {
        // 绑定当前客户唯一标识作为容器ID和消费者组ID，实现隔离
        String containerId = "report-listener-" + customerId;
        ConcurrentMessageListenerContainer<String, String> container = containerFactory.createContainer("testTopic");
        container.getContainerProperties().setGroupId(containerId);
        container.setBeanName(containerId);
        // 绑定当前客户对应的消费逻辑
        container.setupMessageListener((MessageListener<String, String>) record -> {
            // 处理当前客户的报表查询逻辑
            System.out.println("处理客户" + customerId + "的报表请求：" + record.value());
            // 消费完成后可以选择停止销毁当前容器
            container.stop();
        });
        // 启动容器开始消费
        container.start();
        return container;
    }
}

内容的提问来源于stack exchange，提问作者Sumit Sood