Kubernetes环境下Redis Streams多Pod的PEL消息分布不均问题

核心原因分析 #

Redis Stream消费组默认采用**轮询（Round-Robin）**分配消息，但实际出现严重倾斜通常和以下几点相关：

• 消费者单次拉取的消息批次过大，导致单个Pod一次性抢占大量消息，其他Pod无消息可消费
• 不同Pod的消息处理速度差异明显，处理快的Pod会持续拉取下一批次，长期抢占资源
• 消费线程配置不一致，部分Pod线程数更多，抢占消息的效率更高

具体解决方案 #

1. 调小消息拉取批次大小 #

限制每个消费者单次拉取的消息数量，让轮询机制发挥作用，避免单个Pod一次性占坑过多：

@Bean
public StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer(RedisConnectionFactory connectionFactory) {
    StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options =
            StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
                    .batchSize(50) // 根据业务处理速度调整，建议50-200之间
                    .pollTimeout(Duration.ofMillis(100)) // 缩短轮询间隔，确保所有消费者有机会获取消息
                    .build();
    return StreamMessageListenerContainer.create(connectionFactory, options);
}

2. 统一所有Pod的消费线程配置 #

确保所有Pod使用相同的线程池参数，避免因线程数差异导致的抢占能力不均：

@Bean
public TaskExecutor streamTaskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(4); // 所有Pod统一核心线程数
    executor.setMaxPoolSize(8);
    executor.setThreadNamePrefix("redis-stream-");
    executor.initialize();
    return executor;
}

@Bean
public StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer(RedisConnectionFactory connectionFactory, TaskExecutor streamTaskExecutor) {
    StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options =
            StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
                    .executor(streamTaskExecutor)
                    .batchSize(50)
                    .pollTimeout(Duration.ofMillis(100))
                    .build();
    return StreamMessageListenerContainer.create(connectionFactory, options);
}

3. 启用Redis 6.2+的预分配机制（PREFETCH） #

如果你的Redis版本在6.2及以上，利用PREFETCH让Redis更均匀地为在线消费者预分配消息：

StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
        .prefetch(10) // 为每个消费者预分配少量消息，避免单个消费者拉走全部
        .batchSize(50)
        .pollTimeout(Duration.ofMillis(100))
        .build();

4. 排查Kubernetes网络延迟差异 #

个别Pod可能因网络延迟更高，无法及时和Redis交互获取消息。可进入Pod执行redis-cli PING测试响应时间，若存在明显差异，调整Pod调度策略，将所有消费者Pod调度到同一可用区，降低网络延迟差。

5. 自定义消息分配逻辑（极端场景） #

若上述方法无效，可自定义订阅逻辑，手动控制消息拉取范围：

@Bean
public StreamMessageListenerContainer<String, MapRecord<String, String, String>> customStreamContainer(RedisConnectionFactory connectionFactory) throws UnknownHostException {
    StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options =
            StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
                    .batchSize(50)
                    .pollTimeout(Duration.ofMillis(100))
                    .build();
    StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = StreamMessageListenerContainer.create(connectionFactory, options);

    String group = "your-group-name";
    String consumer = "consumer-" + InetAddress.getLocalHost().getHostName();
    StreamOffset<String> offset = StreamOffset.create("your-stream", ReadOffset.lastConsumed());
    StreamReadRequest<String> request = StreamReadRequest.builder(offset)
            .group(group, consumer)
            .autoAcknowledge(false)
            .build();

    container.register(request, message -> {
        // 消息处理逻辑
        // 手动确认：connection.xAck("your-stream", group, message.getId());
    });

    return container;
}

验证方法 #

调整配置后，提交测试消息，通过以下命令检查分配情况：

• XPENDING your-stream your-group：查看各消费者的PEL消息数量
• XINFO CONSUMERS your-stream your-group：查看每个消费者的待处理消息数

内容的提问来源于stack exchange，提问作者Tahmid Bhuiyan