如何为Spring Kafka聚合器配置JdbcMessageStore替代内存存储

当前框架默认通过AggregatorAnnotationPostProcessor直接为AggregatingMessageHandler设置new SimpleMessageStore()，以下是未配置JdbcMessageStore时可正常运行的代码：

@Bean
public ConsumerFactory<?,?> kafkaConsumerFactory(KafkaProperties properties) {
    ConsumerProperties props = properties.buildConsumerProperties();
    return DefaultKafkaConsumerFactory<>(props);
}

@Bean
@InboundChannelAdapter(channel = "fromChannel", poller = @Poller(fixedDelay = "1000"))
public KafkaMessageSource<String, MyPojo> kafkaMessageSource(ConsumerFactory<String, MyPojo> cf) {
    ConsumerProperties props = new ConsumerProperties("topic.in");
    return new KafkaMessageSource<>(cf, props);
}

@Bean
public MessageChannel fromChannel() {
    return new DirectChannel();
}

@Aggregator(inputChannel = "fromChannel", outputChannel = "toChannel")
public List<MyPojo> aggregate(List<MyPojo> list) {
    //apply logic
    return newList;
}

@CorrelationStrategy
public Object correate(Message<MyPojo> message) {
    //apply logic
    //return correlationId; //String
}

@ReleaseStrategy
public boolean checkRelease(Message<MyPojo> message) {
    //apply logic
    //return canRelease; //boolean
}

@Bean
public ProducerFactory<?,?> kafkaProducerFactory(KafkaProperties properties) {
    ConsumerProperties props = properties.buildProducerProperties();
    return DefaultKafkaProducerFactory<>(props);
}

@Bean
@ServiceActivator(inputChannel= "toChannel")
public MessageHandler handler(KafkaTemplate<String, List<MyPojo>> kafkaTemplate) {
    KafkaProducerMessageHandler<String, List<MyPojo>> handler = new KafkaProducerMessageHandler<>(kafkaTemplate);
    handler.setTopicExpression(new LiteralExpression("topic-out"));
    return handler;
}

@Bean
public MessageChannel toChannel() {
    return new DirectChannel();
}

@Bean
public MessageGroupStore messageGroupStore(DataSource dataSource) {
    return new JdbcMessageStore(dataSource);
}

配置步骤 #

• 确保数据源可用
  先确认Spring上下文里已经配置了正确的DataSource Bean，能正常连接你的关系型数据库（比如MySQL、PostgreSQL）。

• 手动创建AggregatingMessageHandler并关联JdbcMessageStore
  默认的注解处理器会自动生成内存版的SimpleMessageStore，所以我们需要手动构建聚合处理器，把自己的JdbcMessageStore注入进去，同时关联聚合、关联、释放逻辑。

  删掉原来的@Aggregator注解方法，替换成下面的Bean配置：

@Bean
public AggregatingMessageHandler aggregatingMessageHandler(MessageGroupStore messageGroupStore,
                                                           CorrelationStrategy correlationStrategy,
                                                           ReleaseStrategy releaseStrategy,
                                                           MessageChannel toChannel) {
    // 这里实现聚合逻辑，对应原来的@Aggregator方法
    MessageGroupProcessor processor = messageGroup -> {
        List<MyPojo> list = messageGroup.getMessages().stream()
                .map(msg -> (MyPojo) msg.getPayload())
                .collect(Collectors.toList());
        // 写入你的聚合逻辑
        return newList;
    };

    AggregatingMessageHandler handler = new AggregatingMessageHandler(processor, messageGroupStore);
    handler.setCorrelationStrategy(correlationStrategy);
    handler.setReleaseStrategy(releaseStrategy);
    handler.setOutputChannel(toChannel);
    handler.setInputChannelName("fromChannel");
    return handler;
}

• 保留策略Bean的有效性
  原来的@CorrelationStrategy和@ReleaseStrategy注解的方法会自动成为Spring Bean，确保这些逻辑能被上面的处理器正常引用。如果需要，也可以把策略逻辑直接内联到处理器配置中。

• 初始化数据库表结构
  JdbcMessageStore依赖特定的表来存储消息分组数据，你需要执行对应的DDL脚本初始化表。核心表包括INT_MESSAGE_GROUP和INT_MESSAGE，以MySQL为例，脚本如下：

CREATE TABLE INT_MESSAGE_GROUP (
    GROUP_ID VARCHAR(255) NOT NULL,
    REGION VARCHAR(100) NOT NULL,
    LAST_UPDATED TIMESTAMP NOT NULL,
    COMPLETE BOOLEAN NOT NULL,
    PRIMARY KEY (GROUP_ID, REGION)
);

CREATE TABLE INT_MESSAGE (
    MESSAGE_ID VARCHAR(255) NOT NULL,
    GROUP_ID VARCHAR(255) NOT NULL,
    REGION VARCHAR(100) NOT NULL,
    CREATED_DATE TIMESTAMP NOT NULL,
    MESSAGE BYTEA NOT NULL,
    PRIMARY KEY (MESSAGE_ID, GROUP_ID, REGION),
    FOREIGN KEY (GROUP_ID, REGION) REFERENCES INT_MESSAGE_GROUP(GROUP_ID, REGION)
);

完整修改后的配置要点 #

• 移除原来的@Aggregator注解方法
• 添加手动构建的AggregatingMessageHandler Bean，注入JdbcMessageStore
• 确保DataSource和JdbcMessageStore Bean正常配置
• 执行数据库表初始化脚本

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