分布式环境下如何在Kafka消费者侧使用Spring Aggregator？

首先明确核心矛盾：Spring Aggregator默认使用本地内存存储聚合中间状态，跨消费者实例场景下状态不互通，同批次消息分散到多个实例时必然会出现永远无法凑齐触发释放的问题。

方案一：无需分布式存储的轻量实现（优先推荐） #

如果你的业务场景允许调整消息投递规则，可以先通过Kafka自身的特性规避分布式聚合的复杂度：

• 给同批次的所有消息设置相同的Kafka消息key，Kafka会默认将相同key的消息路由到同一个分区，而同一消费者组内的一个分区只会被一个消费者实例消费
• 该场景下所有同批次消息都会落到同一个消费者实例，直接使用Spring Aggregator默认的本地SimpleMessageStore即可实现聚合，不需要引入额外的分布式存储组件
• 注意限制：该方案不适合单批次消息量极大、单实例处理性能不足的场景

方案二：分布式聚合实现（适用于多实例处理同批次的场景） #

如果确实需要多个消费者实例并行处理同一批次的消息，必须替换Spring Aggregator的默认存储，此时Hazelcast这类分布式内存数据库是非常合适的选型：

• Spring Integration官方提供了MessageStore接口的标准化扩展，直接引入对应适配依赖即可使用开箱即用的HazelcastMessageStore实现，不需要从零开发适配逻辑
• 核心配置步骤：
  • 所有消息头中必须携带统一的batchId作为聚合的correlationId，Spring Aggregator会基于该字段将同批次消息归为同一聚合组
  • 自定义releaseStrategy：可以将批次总大小n预存入消息头，当聚合组内的消息数量等于n时自动触发释放，生成完成事件；如果批次大小不固定，可以在批次最后一条消息中增加结束标记，收到标记即可触发释放
  • 将HazelcastMessageStore注入到Aggregator组件中，所有聚合的中间状态会存入Hazelcast的共享分布式存储，无论哪个消费者实例收到同批次的消息，都会写入同一个聚合组
• 额外优化配置：给聚合组设置合理的过期时间，避免部分消息丢失导致聚合组长期占用存储资源，过期时可以触发自定义告警处理异常批次

必要的容错处理 #

• 必须做消息幂等校验：Kafka存在消息重复投递的可能，需要在聚合逻辑中基于消息唯一ID去重，避免重复计数导致聚合组提前释放
• 建议配置死信队列：聚合超时的残次批次消息可以转入死信队列，后续人工介入或者自动重试处理

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