springboot redis stream批量消费

Spring Boot提供了`RedisMessageListenerContainer`来监听Redis Stream中的消息，并提供了`StreamMessageListenerContainer.MessageListenerContainer`接口来处理消息。可以使用`RedisTemplate`的`execute(RedisCallback<T> action, boolean exposeConnection)`方法来批量处理Redis Stream消息。

以下是一个示例代码，演示如何使用`RedisMessageListenerContainer`和`execute()`方法来批量消费Redis Stream中的消息：

@Component
public class StreamMessageListener {

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    @Autowired
    private RedisMessageListenerContainer redisMessageListenerContainer;

    @PostConstruct
    public void init() {
        StreamMessageListenerContainer<Message<String, String>> container = StreamMessageListenerContainer
                .create(redisTemplate.getConnectionFactory());
        container.receiveAutoAck("stream-name", Duration.ofSeconds(1), new BatchMessageListener());
        redisMessageListenerContainer.addMessageListener(container);
    }

    private class BatchMessageListener implements StreamMessageListenerContainer.MessageListener<String, String> {
        private List<MapRecord<String, String, String>> buffer = new ArrayList<>();
        private final int batchSize = 10;

        @Override
        public void onMessage(Message<String, String> message) {
            buffer.add((MapRecord<String, String, String>) message);
            if (buffer.size() >= batchSize) {
                redisTemplate.execute((RedisCallback<Void>) connection -> {
                    for (MapRecord<String, String, String> record : buffer) {
                        // 批量消费消息
                        // your business logic here
                    }
                    return null;
                }, true);
                buffer.clear();
            }
        }
    }
}

在`init()`方法中，我们创建了一个`StreamMessageListenerContainer`实例，并指定了要监听的Redis Stream的名称、轮询间隔和消息处理器，然后将其添加到`RedisMessageListenerContainer`中。

在`BatchMessageListener`中，我们创建了一个缓冲区`buffer`来存储消息，当缓冲区中的消息数量达到`batchSize`时，我们使用`execute()`方法批量消费消息。在`execute()`方法中，我们可以在`RedisCallback`的回调函数中执行自己的业务逻辑。

需要注意的是，使用`execute()`方法批量消费Redis Stream消息时，需要将`exposeConnection`参数设置为`true`，这样可以让Spring Boot自动管理连接生命周期，确保连接的正确关闭。