基于Redis Streams的實時消息處理實戰(zhàn)指南

更新時間：2025年07月16日 09:12:39 作者：淺沫云歸

這篇文章主要為大家詳細介紹了在生產環(huán)境中基于 Redis Streams 構建實時消息處理的完整經驗,包括技術選型、核心代碼示例、踩坑解決和優(yōu)化方案,希望對大家有所幫助

業(yè)務場景描述

在我們公司的電商平臺中，存在大量異步事件需要實時處理，例如用戶下單、庫存更新、支付回調等。這些事件對消息的可靠性、順序性和高吞吐量有較高要求。傳統(tǒng)的消息中間件（如Kafka、RabbitMQ）在運維成本或部署復雜度上存在一定挑戰(zhàn)，在部分場景下難以滿足“輕量、低延遲、易集成” 的需求。

經過調研和驗證，Redis 6.0+ 提供的 Streams 特性在嵌入式部署、快速上手方面具有顯著優(yōu)勢。本篇文章將分享我們在生產環(huán)境中基于 Redis Streams 構建實時消息處理的完整經驗，包括技術選型、核心代碼示例、踩坑解決和優(yōu)化方案。

技術選型過程

消息可靠性：Redis Streams 支持持久化，且提供 ACK 機制和 Pending List，能夠有效追蹤消費進度。
順序消費：同一消費者組內，可保證分片流（同一 key）中消息按寫入順序被串行消費。
橫向擴展：可通過 Stream 分片（多個 Stream Key）或消費者組內多實例并行消費提高吞吐。
運營成本：Redis 已是團隊基礎設施，集群部署與監(jiān)控成熟度高，二次成本低。
客戶端生態(tài)：Lettuce、Jedis、Redisson 等客戶端均有支持，編碼友好。

基于以上考量，最終選型 Redis Streams，落地于現(xiàn)有 Redis 集群，無需額外獨立中間件部署。

實現(xiàn)方案詳解

環(huán)境與依賴

Maven 依賴（以 Lettuce 客戶端為例）：

<dependencies>
    <dependency>
        <groupId>io.lettuce</groupId>
        <artifactId>lettuce-core</artifactId>
        <version>6.1.5.RELEASE</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-api</artifactId>
        <version>1.7.30</version>
    </dependency>
    <dependency>
        <groupId>ch.qos.logback</groupId>
        <artifactId>logback-classic</artifactId>
        <version>1.2.3</version>
    </dependency>
</dependencies>

SpringBoot 配置（application.yml）：

spring:
  redis:
    host: redis-cluster-host
    port: 6379
    password: your_password
    timeout: 2000ms

流程設計

Producer 將事件寫入 Stream：XADD
多消費者（Consumer Group）并行讀?。篨READGROUP
消費確認：XACK
異常消息追蹤：Pending-List 與 XCLAIM 回補處理

生產者實現(xiàn)

import io.lettuce.core.RedisClient;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.sync.RedisCommands;
import java.util.HashMap;
import java.util.Map;

public class RedisStreamProducer {
    private RedisClient client;
    private StatefulRedisConnection<String, String> connection;
    private RedisCommands<String, String> commands;
    private static final String STREAM_KEY = "orderStream";

    public RedisStreamProducer(String uri) {
        client = RedisClient.create(uri);
        connection = client.connect();
        commands = connection.sync();
    }

    public String sendMessage(Map<String, String> message) {
        // XADD key * field value [field value ...]
        return commands.xadd(STREAM_KEY, message);
    }

    public void shutdown() {
        connection.close();
        client.shutdown();
    }

    public static void main(String[] args) {
        RedisStreamProducer producer = new RedisStreamProducer("redis://:your_password@redis-host:6379/0");
        Map<String, String> order = new HashMap<>();
        order.put("orderId", "123456");
        order.put("userId", "u7890");
        order.put("amount", "258.50");
        String messageId = producer.sendMessage(order);
        System.out.println("消息發(fā)送成功, ID=" + messageId);
        producer.shutdown();
    }
}

消費者實現(xiàn)

import io.lettuce.core.RedisClient;
import io.lettuce.core.StreamMessage;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.sync.RedisCommands;
import io.lettuce.core.models.stream.Consumer;
import io.lettuce.core.models.stream.PendingMessage;

import java.time.Duration;
import java.util.List;
import java.util.Map;

public class RedisStreamConsumer {
    private RedisClient client;
    private StatefulRedisConnection<String, String> connection;
    private RedisCommands<String, String> commands;

    private static final String STREAM_KEY = "orderStream";
    private static final String GROUP_NAME = "orderGroup";
    private static final String CONSUMER_NAME = "consumer-1";

    public RedisStreamConsumer(String uri) {
        client = RedisClient.create(uri);
        connection = client.connect();
        commands = connection.sync();
        // 創(chuàng)建消費者組, 如果已創(chuàng)建可 ignore
        try {
            commands.xgroupCreate(STREAM_KEY, GROUP_NAME, "$", true);
        } catch (Exception e) {
            // Group exists
        }
    }

    public void consume() {
        while (true) {
            // 從 Pending List 先處理未 ack 的消息
            List<PendingMessage> pending = commands.xpending(STREAM_KEY, GROUP_NAME, Range.unbounded(), Limit.from(10));
            for (PendingMessage pm : pending) {
                // 重新消費
                StreamMessage<String, String> msg = commands.xclaim(
                    STREAM_KEY,
                    GROUP_NAME,
                    CONSUMER_NAME,
                    5000,
                    pm.getId());
                process(msg.getBody());
                commands.xack(STREAM_KEY, GROUP_NAME, pm.getId());
            }

            // 正常讀取新消息
            List<StreamMessage<String, String>> messages = commands.xreadgroup(
                Consumer.from(GROUP_NAME, CONSUMER_NAME),
                XReadArgs.StreamOffset.lastConsumed(STREAM_KEY));
            if (messages != null) {
                for (StreamMessage<String, String> msg : messages) {
                    process(msg.getBody());
                    commands.xack(STREAM_KEY, GROUP_NAME, msg.getId());
                }
            }

            // 輪詢間隔
            try {
                Thread.sleep(200);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
    }

    private void process(Map<String, String> body) {
        // 業(yè)務處理邏輯
        System.out.println("處理訂單: " + body);
    }

    public void shutdown() {
        connection.close();
        client.shutdown();
    }

    public static void main(String[] args) {
        RedisStreamConsumer consumer = new RedisStreamConsumer("redis://:your_password@redis-host:6379/0");
        consumer.consume();
        consumer.shutdown();
    }
}