Sping Cloud Stream Kafka-以批处理模式消费消息，并作为单个处理消息发送出去

Question

我试着为下一个spring cloud流版本准备我们的应用程序。（当前使用3.0.0.rc1)。用Kafka的活页夹。

现在我们收到一个消息，处理它，并将它重新发送到另一个主题。单独处理每个消息会导致对数据库的大量单个请求。

在3.0.0版本中，我们希望以批处理的方式处理消息，这样我们就可以在批更新中保存数据。

在当前版本中，我们使用了@enablebinding、@streamlistener

@StreamListener( ExchangeableItemProcessor.STOCK_INPUT )
public void processExchangeableStocks( final ExchangeableStock item ) {
    publishItems( exchangeableItemProcessor.stocks(), articleService.updateStockInformation( Collections.singletonList( item ) ) );
}

void publishItems( final MessageChannel messageChannel, final List<? extends ExchangeableItem> item ) {
    for ( final ExchangeableItem exchangeableItem : item ) {
        final Message<ExchangeableItem> message = MessageBuilder.withPayload( exchangeableItem )
                            .setHeader( "partitionKey", exchangeableItem.getId() )
                            .build();
        messageChannel.send( message )
    }
}

我已将使用者属性设置为“批处理模式”，并将签名更改为list<>，但这样做会导致接收到list 而不是预期的list 。Ofc可以在之后进行转换，但这感觉像是“meh”，我认为这是应该在监听器被调用之前发生的事情。

然后我尝试了（新的）函数版本，使用效果很好。我也喜欢这个简单的处理版本

@Bean
public Function<List<ExchangeableStock>, List<ExchangeableStock>> stocks() {
    return articleService::updateStockInformation;
}

但输出主题现在将对象列表作为一条消息接收,并且随后的使用者无法正常工作。

我想我错过了什么...

我是否需要添加某种MessageConverter（对于注释驱动的版本），或者是否有一种方法可以在函数版本中实现所需的行为？

Answer 1

IIRC，批处理模式只支持函数。

可以不使用consumer > 并像当前在StreamListener中所做的那样将消息发送到通道吗？