Spring中Kotlin协程Channel后台处理及Java多生产者多消费者方案

一、Kotlin + Spring 协程Channel消费实现 #

需求背景 #

有一个长轮询机器人，其onMessage(callbackSuspendAction: suspend.() -> Unit)回调是耗时20-30秒的重计算任务。目前将任务封装为懒启动协程存入messageHandlersQueue: Channel<Job>，代码如下：

messageHandlersQueue.send(
    coroutineScope {
        launch(start = CoroutineStart.LAZY) {
            val vkClient = VkClient(actor.userActor)
            processVkUserQuery(msg.chat.id, addressName, vkClient)
        }
    }
)

希望借助Spring生命周期启动5个后台协程消费者，替代GlobalScope的写法，需要类似@Scheduled(fixedDelay=0)或支持suspend的@PostConstruct机制。

实现方案 #

Spring本身不直接支持suspend版的@PostConstruct，但可以通过@PostConstruct启动Spring管理的协程作用域，在其中启动多个消费协程：

@Component
class MessageHandlerConsumer(
    private val messageHandlersQueue: Channel<Job>,
    private val coroutineScope: CoroutineScope // 注入Spring管理的协程作用域，需引入kotlinx-coroutines-spring依赖
) {

    @PostConstruct
    fun startConsumers() {
        repeat(5) {
            coroutineScope.launch {
                setupMessageHandlerConsumer()
            }
        }
    }

    private suspend fun setupMessageHandlerConsumer() {
        for (job in messageHandlersQueue) {
            job.start() // 启动懒加载的协程任务
            job.join() // 可选：等待任务完成再处理下一个
        }
    }
}

注意：引入kotlinx-coroutines-spring依赖后，Spring会自动管理协程作用域，避免GlobalScope带来的生命周期不一致问题。

二、临时替代方案（BlockingQueue + 线程池） #

暂未找到合适的Channel解决方案时，可切换为BlockingQueue配合线程池实现，代码如下：

生产者逻辑（onMessage中提交任务） #

WAITING_SEARCH_INPUT -> {
    if (validateAddressLink(text)) {
        val addressName = extractAddressName(text)
        val offered = searchQueries.offer(VkUserQuery(chatId, addressName, session.client))
        if (!offered) {
            bot.sendMessage(chatId, TOO_MUCH_LOAD_MESSAGE)
        }
    } else {
        bot.sendMessage(chatId, WRONG_ADDRESS_NAME_FORMAT_MESSAGE)
    }
}

消费者启动与处理逻辑 #

private fun setupSearchQueriesProcessor() {
    for (i in 0 until RAKING_THREADS_COUNT) {
        executorService.submit {
            searchQueryConsumer()
        }
    }
}

private fun searchQueryConsumer() {
    while (true) {
        val searchQuery = searchQueries.take()
        vkSessionRegistry.setState(searchQuery.chatId, SEARCHING)
        processVkUserQuery(
            searchQuery.chatId,
            searchQuery.addressName,
            searchQuery.client
        )
        vkSessionRegistry.setState(searchQuery.chatId, RESTING)
    }
}

三、纯Spring Java（无Kafka、Kotlin）多生产者多消费者最优方案 #

直接使用BlockingQueue配合Spring管理的线程池（ThreadPoolTaskExecutor）即可，无需额外中间件：

1. 配置线程池 #

@Configuration
public class ThreadPoolConfig {
    @Bean
    public ThreadPoolTaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5); // 核心线程数
        executor.setMaxPoolSize(10); // 最大线程数
        executor.setQueueCapacity(100); // 队列容量，超过则触发拒绝策略
        executor.setThreadNamePrefix("query-consumer-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); // 拒绝时让提交者执行任务
        executor.initialize();
        return executor;
    }
}

2. 定义任务类 #

public class VkUserQuery {
    private Long chatId;
    private String addressName;
    private VkClient client;

    // 构造器、getter、setter
}

3. 生产者逻辑 #

// 在onMessage回调中
if (validateAddressLink(text)) {
    String addressName = extractAddressName(text);
    VkUserQuery query = new VkUserQuery(chatId, addressName, session.getClient());
    boolean offered = searchQueries.offer(query);
    if (!offered) {
        bot.sendMessage(chatId, TOO_MUCH_LOAD_MESSAGE);
    }
} else {
    bot.sendMessage(chatId, WRONG_ADDRESS_NAME_FORMAT_MESSAGE);
}

4. 消费者启动与处理 #

@Component
public class SearchQueryConsumer {
    private final BlockingQueue<VkUserQuery> searchQueries;
    private final ThreadPoolTaskExecutor taskExecutor;
    private final VkSessionRegistry vkSessionRegistry;

    @Autowired
    public SearchQueryConsumer(BlockingQueue<VkUserQuery> searchQueries, ThreadPoolTaskExecutor taskExecutor, VkSessionRegistry vkSessionRegistry) {
        this.searchQueries = searchQueries;
        this.taskExecutor = taskExecutor;
        this.vkSessionRegistry = vkSessionRegistry;
    }

    @PostConstruct
    public void startConsumers() {
        for (int i = 0; i < 5; i++) {
            taskExecutor.execute(this::consume);
        }
    }

    private void consume() {
        while (!Thread.currentThread().isInterrupted()) {
            try {
                VkUserQuery query = searchQueries.take();
                vkSessionRegistry.setState(query.getChatId(), "SEARCHING");
                processVkUserQuery(query.getChatId(), query.getAddressName(), query.getClient());
                vkSessionRegistry.setState(query.getChatId(), "RESTING");
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt(); // 恢复中断状态
                break;
            }
        }
    }

    private void processVkUserQuery(Long chatId, String addressName, VkClient client) {
        // 耗时20-30秒的重计算逻辑
    }
}

关键说明 #

• 用Spring管理的ThreadPoolTaskExecutor统一配置线程参数，便于生命周期管理
• BlockingQueue的take()方法会自动阻塞等待任务，无需轮询
• 处理中断异常，保证应用优雅停机
• 拒绝策略可根据业务需求调整（如丢弃任务、记录告警日志等）

内容的提问来源于stack exchange，提问作者Dmitrii Dulaev