1
import lombok.extern.slf4j.Slf4j; // 新增导入
2
import org.springframework.beans.factory.annotation.Autowired;
3
import org.springframework.data.redis.connection.stream.MapRecord;
4
import org.springframework.data.redis.connection.stream.RecordId;
5
import org.springframework.data.redis.core.StringRedisTemplate;
6
import org.springframework.stereotype.Service;
7

8
import java.util.HashMap;
9
import java.util.Map;
10
// com.alibaba.fastjson.JSON 如果需要，请确保此依赖存在
11
// import com.alibaba.fastjson.JSON;
12

13
@Service
14
@Slf4j // 新增注解
15
public class RedisStreamProducer {
16

17
    @Autowired
18
    private StringRedisTemplate stringRedisTemplate;
19

20
    /**
21
     * 发送消息到Stream队列
22
     * @param streamKey Stream队列的key
23
     * @param messageData 消息数据
24
     * @return 消息ID
25
     */
26
    public String sendMessage(String streamKey, Map<String, Object> messageData) {
27
        try {
28
            // 将消息发送到Stream队列
29
            RecordId recordId = stringRedisTemplate.opsForStream()
30
                    .add(streamKey, messageData);
31
            log.info("消息发送成功，消息ID: {}", recordId.getValue());
32
            return recordId.getValue();
33
        } catch (Exception e) {
34
            log.error("发送消息失败: {}", e.getMessage());
35
            throw new RuntimeException("消息发送失败", e);
36
        }
37
    }
38

39
    /**
40
     * 发送业务消息的示例
41
     */
42
    public void sendBusinessMessage(Object businessData) {
43
        Map<String, Object> messageMap = new HashMap<>();
44
        messageMap.put("businessId", businessData.getId());
45
        messageMap.put("businessType", businessData.getClass().getSimpleName());
46
        messageMap.put("data", JSON.toJSONString(businessData));
47
        messageMap.put("timestamp", System.currentTimeMillis());
48

49
        sendMessage("stream:business-queue", messageMap);
50
    }
51
}

1
@Component
2
@Slf4j
3
public class RedisStreamConsumer implements Runnable {
4

5
    @Autowired
6
    private StringRedisTemplate stringRedisTemplate;
7

8
    // 配置信息
9
    private static final String STREAM_KEY = "stream:business-queue";
10
    private static final String GROUP_NAME = "business-group";
11
    private static final String CONSUMER_NAME = "consumer-1";
12

13
    @PostConstruct
14
    public void init() {
15
        // 启动时创建消费者组（如果不存在）
16
        try {
17
            stringRedisTemplate.opsForStream()
18
                    .createGroup(STREAM_KEY, GROUP_NAME, "0");
19
        } catch (Exception e) {
20
            // 消费者组已存在，忽略异常
21
            log.info("消费者组可能已存在: {}", e.getMessage());
22
        }
23

24
        // 启动消费者线程
25
        new Thread(this).start();
26
    }
27

28
    @Override
29
    public void run() {
30
        while (true) {
31
            try {
32
                // 从Stream中获取消息
33
                List<MapRecord<String, Object, Object>> messages = stringRedisTemplate.opsForStream()
34
                        .read(Consumer.from(GROUP_NAME, CONSUMER_NAME),
35
                              StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
36
                              StreamOffset.create(STREAM_KEY, ReadOffset.lastConsumed()));
37

38
                // 没有获取到消息，继续循环
39
                if (messages == null || messages.isEmpty()) {
40
                    continue;
41
                }
42

43
                // 处理消息
44
                for (MapRecord<String, Object, Object> message : messages) {
45
                    processMessage(message);
46
                }
47

48
            } catch (Exception e) {
49
                log.error("处理消息异常", e);
50
                handlePendingList();
51
            }
52
        }
53
    }
54

55
    /**
56
     * 处理单个消息
57
     */
58
    private void processMessage(MapRecord<String, Object, Object> message) {
59
        try {
60
            Map<Object, Object> messageData = message.getValue();
61

62
            // 提取消息内容
63
            String businessId = (String) messageData.get("businessId");
64
            String businessType = (String) messageData.get("businessType");
65
            String data = (String) messageData.get("data");
66

67
            log.info("开始处理消息 - ID: {}, 业务类型: {}", businessId, businessType);
68

69
            // 执行具体的业务逻辑（这里可以根据businessType分发到不同的处理器）
70
            handleBusinessLogic(businessType, data);
71

72
            // 手动确认消息
73
            stringRedisTemplate.opsForStream()
74
                    .acknowledge(STREAM_KEY, GROUP_NAME, message.getId());
75

76
            log.info("消息处理完成 - ID: {}", businessId);
77

78
        } catch (Exception e) {
79
            log.error("处理消息失败: {}", e.getMessage(), e);
80
            // 这里可以根据业务需要决定是否重试或丢弃消息
81
        }
82
    }
83

84
    /**
85
     * 执行具体的业务逻辑
86
     * 这里可以根据业务类型分发到不同的处理器
87
     */
88
    private void handleBusinessLogic(String businessType, String data) {
89
        switch (businessType) {
90
            case "Order":
91
                handleOrderLogic(data);
92
                break;
93
            case "Payment":
94
                handlePaymentLogic(data);
95
                break;
96
            case "Notification":
97
                handleNotificationLogic(data);
98
                break;
99
            default:
100
                log.warn("未知的业务类型: {}", businessType);
101
        }
102
    }
103

104
    private void handleOrderLogic(String data) {
105
        // 订单相关业务逻辑
106
        log.info("处理订单业务: {}", data);
107
        // 实际业务处理...
108
    }
109

110
    private void handlePaymentLogic(String data) {
111
        // 支付相关业务逻辑
112
        log.info("处理支付业务: {}", data);
113
        // 实际业务处理...
114
    }
115

116
    private void handleNotificationLogic(String data) {
117
        // 通知相关业务逻辑
118
        log.info("处理通知业务: {}", data);
119
        // 实际业务处理...
120
    }
121

122
    /**
123
     * 处理Pending List中的消息
124
     * 当消费者处理消息异常时，处理未确认的消息
125
     */
126
    private void handlePendingList() {
127
        while (true) {
128
            try {
129
                // 获取Pending List中的消息
130
                List<MapRecord<String, Object, Object>> pendingMessages = stringRedisTemplate.opsForStream()
131
                        .read(Consumer.from(GROUP_NAME, CONSUMER_NAME),
132
                              StreamReadOptions.empty().count(1),
133
                              StreamOffset.create(STREAM_KEY, ReadOffset.from("0")));
134

135
                if (pendingMessages != null && !pendingMessages.isEmpty()) {
136
                    for (MapRecord<String, Object, Object> message : pendingMessages) {
137
                        processMessage(message);
138
                    }
139
                } else {
140
                    // 没有更多pending消息，退出循环
141
                    break;
142
                }
143
            } catch (Exception e) {
144
                log.error("处理Pending List异常", e);
145
                break;
146
            }
147
        }
148
    }
149
}