springboot redis stream批量消费

在 Spring Boot 中，可以使用 RedisTemplate 和 StreamMessageListenerContainer 来实现批量消费 Redis Stream。

首先，需要创建一个 RedisTemplate 实例并配置连接信息：

@Configuration
public class RedisConfig {

    @Bean
    public RedisConnectionFactory redisConnectionFactory() {
        return new LettuceConnectionFactory("localhost", 6379);
    }

    @Bean
    public RedisTemplate<String, Object> redisTemplate() {
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisConnectionFactory());
        redisTemplate.setDefaultSerializer(new GenericJackson2JsonRedisSerializer());
        return redisTemplate;
    }

}

然后，创建一个 StreamMessageListenerContainer 实例并配置监听器：

@Component
public class StreamConsumer {

    private final RedisTemplate<String, Object> redisTemplate;

    public StreamConsumer(RedisTemplate<String, Object> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    @Bean
    public StreamMessageListenerContainer<String, MapRecord<String, String, Object>> streamMessageListenerContainer() {
        StreamMessageListenerContainer<String, MapRecord<String, String, Object>> container =
                StreamMessageListenerContainer.create(redisTemplate.getConnectionFactory());
        container.setAutoStartup(true);

        StreamMessageListener<String, MapRecord<String, String, Object>> listener =
                new StreamMessageListener<String, MapRecord<String, String, Object>>() {
                    @Override
                    public void onMessage(MapRecord<String, String, Object> message) {
                        // 处理消息
                    }
                };

        container.setupMessageListener(listener, StreamOffset.fromStart("stream-name"));
        container.start();

        return container;
    }

}

其中，StreamOffset.fromStart("stream-name") 表示从头开始消费指定名字的 Stream。

最后，在 onMessage 方法中处理消息即可：

@Override
public void onMessage(MapRecord<String, String, Object> message) {
    List<MapRecord<String, String, Object>> messages = new ArrayList<>();
    messages.add(message);

    // 批量处理消息
    while (true) {
        MapRecord<String, String, Object> nextMessage = container.getStreamOperations()
                .poll(Collections.singletonMap("stream-name", "consumer-group"), Duration.ZERO, messages.size() - 1);
        if (nextMessage == null) {
            break;
        }
        messages.add(nextMessage);
    }

    // 处理消息
    for (MapRecord<String, String, Object> msg : messages) {
        // ...
    }
}

在这个例子中，我们使用了 poll 方法从 Redis Stream 中批量获取消息，并通过 while 循环来不断获取下一批消息，直到获取完所有消息。然后，再遍历所有消息进行处理。