基于Redis Streams的实时消息处理实战指南

<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>

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

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("消息发送成功, ID=" + messageId);
        producer.shutdown();
    }
}

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();
        // 创建消费者组, 如果已创建可 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) {
        // 业务处理逻辑
        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();
    }
}