将JdbcMessageStore添加到Spring Kafka聚合器

7vhp5slm  于 2023-10-15  发布在  Apache
关注(0)|答案(1)|浏览(104)

如何将JdbcMessageStore设置为Aggregator,使其使用RDBMS而不是内存中的消息存储?目前AggregatorAnnotationPostProcessor通过框架直接将new SimpleMessageStore()设置为AggregatingMessageHandler
这里是配置,这是工作的预期没有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);
}
nuypyhwy

nuypyhwy1#

为此,我们建议通过AggregatorFactoryBean使用更高级的配置:https://docs.spring.io/spring-integration/reference/aggregator.html#aggregator-annotations

@Bean
    @ServiceActivator(inputChannel = "fromChannel")
    AggregatorFactoryBean aggregatorFactoryBean(MessageGroupStore messageGroupStore, MessageChannel toChannel) {
        AggregatorFactoryBean aggregatorFactoryBean = new AggregatorFactoryBean();
        aggregatorFactoryBean.setMessageStore(messageGroupStore);
        aggregatorFactoryBean.setProcessorBean(...);
        aggregatorFactoryBean.setCorrelationStrategy(...);
        aggregatorFactoryBean.setReleaseStrategy(...);
        aggregatorFactoryBean.setOutputChannel(toChannel);
        return aggregatorFactoryBean;
    }

相关问题