Spring Boot Redis Stream批量消费

在Spring Boot中，我们可以使用`@StreamListener`注解来监听Redis Stream中的消息。但是默认情况下，每次只能消费一条消息。如果需要批量消费Redis Stream中的消息，可以使用`RedisTemplate`的`xread()`方法来实现。

以下是一个示例代码，展示如何使用Spring Boot Redis Stream批量消费：

@Component
public class RedisStreamConsumer {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @StreamListener("input-stream")
    public void consume(StreamRecord<String, Object> record) {
        // 处理单条消息的逻辑
    }

    @Scheduled(fixedDelay = 1000)
    public void batchConsume() {
        StreamOperations<String, Object, Object> streamOperations = redisTemplate.opsForStream();
        StreamReadOptions readOptions = StreamReadOptions.empty().block(Duration.ZERO).count(10);
        List<MapRecord<String, Object, Object>> records = streamOperations.read(readOptions, StreamOffset.create("input-stream", ReadOffset.lastConsumed()));
        records.forEach(record -> {
            // 处理批量消息的逻辑
        });
    }
}

在上面的代码中，`@Scheduled`注解指定了定时任务的执行间隔（本例中为1秒）。`batchConsume()`方法使用`RedisTemplate`的`opsForStream()`方法获取`StreamOperations`对象，并使用`StreamReadOptions`对象指定了读取的消息数量（本例中为10）。

`read()`方法返回一个`List`，其中包含了读取到的一批消息。我们可以遍历这个`List`，并在其中处理每条消息的逻辑。