快捷導(dǎo)航

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

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

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

業(yè)務(wù)場景描述

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

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

技術(shù)選型過程

消息可靠性：Redis Streams 支持持久化，且提供 ACK 機(jī)制和 Pending List，能夠有效追蹤消費進(jìn)度。
順序消費：同一消費者組內(nèi)，可保證分片流（同一 key）中消息按寫入順序被串行消費。
橫向擴(kuò)展：可通過 Stream 分片（多個 Stream Key）或消費者組內(nèi)多實例并行消費提高吞吐。
運營成本：Redis 已是團(tuán)隊基礎(chǔ)設(shè)施，集群部署與監(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

流程設(shè)計

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

生產(chǎn)者實現(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è)務(wù)處理邏輯
        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();
    }
}