SpringBoot Kafka消费者按matchId串行/并行处理消息方案问询

方案一：自定义分组任务调度器 #

通过维护每个matchId对应的任务队列，确保同一matchId的任务串行执行，不同matchId的任务由线程池并行处理，适合不依赖Kafka分区配置的场景。

核心思路：用ConcurrentHashMap存储每个matchId对应的阻塞队列，提交任务时将任务放入对应队列，同时启动线程消费该队列（仅当队列未被消费时），队列空后自动清理避免内存泄漏。

代码示例：

@Component
public class MatchTaskManager {
    private static final Logger log = LoggerFactory.getLogger(MatchTaskManager.class);
    private final ConcurrentHashMap<String, BlockingQueue<Runnable>> matchTaskQueues = new ConcurrentHashMap<>();
    // 根据业务并发需求调整线程池大小
    private final ExecutorService executor = Executors.newFixedThreadPool(10);

    public void submitTask(String matchId, Runnable task) {
        // 获取或创建当前matchId的任务队列
        BlockingQueue<Runnable> taskQueue = matchTaskQueues.computeIfAbsent(matchId, k -> new LinkedBlockingQueue<>());
        taskQueue.add(task);

        // 启动队列消费线程（仅当队列未被处理时）
        executor.submit(() -> {
            try {
                while (!Thread.currentThread().isInterrupted()) {
                    Runnable runnable = taskQueue.take();
                    try {
                        runnable.run();
                    } catch (Exception e) {
                        log.error("处理matchId [{}] 的任务失败", matchId, e);
                    }
                    // 队列空时移除，释放内存
                    if (taskQueue.isEmpty()) {
                        matchTaskQueues.remove(matchId);
                        break;
                    }
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
    }
}

在Kafka消费者中调用：

@KafkaListener(topics = "matchupdates")
public void consumeMatchUpdate(ConsumerRecord<String, String> record) {
    String payload = record.value();
    // 从payload中解析matchId，自行实现解析逻辑
    String matchId = extractMatchId(payload);
    
    Runnable dbTask = () -> {
        // 处理消息并写入数据库的业务逻辑
        processAndSaveToDatabase(payload);
    };
    
    matchTaskManager.submitTask(matchId, dbTask);
}

方案二：利用Kafka分区与Key路由 #

如果消息发送时将matchId作为Kafka消息的Key，Kafka会自动将相同Key的消息分配到同一分区。此时配置消费者的并发数等于Topic分区数，每个分区对应一个消费线程，天然保证同一matchId的消息串行处理，不同matchId的消息并行处理。

配置要点：

1. 发送消息时指定matchId为Key：

kafkaTemplate.send("matchupdates", matchId, payload);

2. Spring Boot配置文件中设置消费者并发数等于Topic分区数：

spring:
  kafka:
    consumer:
      group-id: match-update-consumer-group
      concurrency: 10 # 需与Topic的分区数一致
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

消费者代码：

@KafkaListener(topics = "matchupdates")
public void consumeMatchUpdates(List<ConsumerRecord<String, String>> records) {
    // 同一分区的消息按顺序到达，串行处理即可
    for (ConsumerRecord<String, String> record : records) {
        String payload = record.value();
        processAndSaveToDatabase(payload);
    }
}

该方案无需额外维护队列，依赖Kafka自身的分区机制，适合能控制消息发送逻辑的场景。

方案三：响应式编程分组处理（Reactor） #

如果项目使用Spring WebFlux或Reactor，可以将消息流按matchId分组，每个分组串行处理，不同分组由弹性线程池并行执行。

代码示例：

@KafkaListener(topics = "matchupdates", containerFactory = "reactiveKafkaListenerContainerFactory")
public Flux<Void> consumeMatchUpdate(Flux<ConsumerRecord<String, String>> recordFlux) {
    return recordFlux
            // 按matchId分组
            .groupBy(record -> extractMatchId(record.value()))
            // 每个分组内串行处理，不同分组并行
            .flatMap(group -> group.concatMap(record -> 
                Mono.fromRunnable(() -> processAndSaveToDatabase(record.value()))
                    .subscribeOn(Schedulers.boundedElastic())
            ));
}

该方案代码简洁，适合响应式架构，自动处理线程调度与分组逻辑。

注意事项 #

• 异常处理：需捕获任务执行异常，避免单个任务失败导致整个消费线程终止。
• 内存泄漏：方案一中需确保空队列被及时移除，避免长期未更新的matchId占用内存。
• 并行度控制：线程池大小或Kafka分区数需根据业务并发量调整，避免资源过载。

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