From 808dff53d26d48b2c8613f260ef76d4246892b74 Mon Sep 17 00:00:00 2001
From: Azure <983547216@qq.com>
Date: Mon, 22 Jun 2026 13:42:37 +0800
Subject: [PATCH] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0=20RedisSseListener?=
 =?UTF-8?q?=20=E5=92=8C=20SseConfig=20Redis=20Pub/Sub=20=E9=85=8D=E7=BD=AE?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../webgamebackend/config/SseConfig.java      | 28 +++++++++++
 .../webgamebackend/sse/RedisSseListener.java  | 47 +++++++++++++++++++
 2 files changed, 75 insertions(+)
 create mode 100644 src/main/java/com/webgame/webgamebackend/config/SseConfig.java
 create mode 100644 src/main/java/com/webgame/webgamebackend/sse/RedisSseListener.java

diff --git a/src/main/java/com/webgame/webgamebackend/config/SseConfig.java b/src/main/java/com/webgame/webgamebackend/config/SseConfig.java
new file mode 100644
index 0000000..02fdc3f
--- /dev/null
+++ b/src/main/java/com/webgame/webgamebackend/config/SseConfig.java
@@ -0,0 +1,28 @@
+package com.webgame.webgamebackend.config;
+
+import com.webgame.webgamebackend.common.event.SseEventPublisher;
+import com.webgame.webgamebackend.sse.RedisSseListener;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.listener.ChannelTopic;
+import org.springframework.data.redis.listener.RedisMessageListenerContainer;
+
+/**
+ * SSE 相关 Bean 配置
+ *
+ * 注册 Redis Pub/Sub 消息监听器，订阅频道 "sse:events"。
+ */
+@Configuration
+public class SseConfig {
+
+    @Bean
+    public RedisMessageListenerContainer sseMessageListenerContainer(
+            RedisConnectionFactory connectionFactory,
+            RedisSseListener listener) {
+        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
+        container.setConnectionFactory(connectionFactory);
+        container.addMessageListener(listener, new ChannelTopic(SseEventPublisher.CHANNEL));
+        return container;
+    }
+}
diff --git a/src/main/java/com/webgame/webgamebackend/sse/RedisSseListener.java b/src/main/java/com/webgame/webgamebackend/sse/RedisSseListener.java
new file mode 100644
index 0000000..0c607fa
--- /dev/null
+++ b/src/main/java/com/webgame/webgamebackend/sse/RedisSseListener.java
@@ -0,0 +1,47 @@
+package com.webgame.webgamebackend.sse;
+
+import com.alibaba.fastjson2.JSON;
+import com.webgame.webgamebackend.common.event.SseEvent;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.connection.Message;
+import org.springframework.data.redis.connection.MessageListener;
+import org.springframework.stereotype.Component;
+
+import java.nio.charset.StandardCharsets;
+
+/**
+ * Redis Pub/Sub 消息监听器
+ *
+ * 接收来自 Redis 频道 "sse:events" 的消息，
+ * 反序列化后根据事件类型（广播/定向）转发给本地 SseConnectionManager。
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class RedisSseListener implements MessageListener {
+
+    private final SseConnectionManager connectionManager;
+
+    @Override
+    public void onMessage(Message message, byte[] pattern) {
+        String body = new String(message.getBody(), StandardCharsets.UTF_8);
+        log.debug("[SSE] 收到 Redis 消息, channel={}", new String(message.getChannel(), StandardCharsets.UTF_8));
+
+        try {
+            SseEvent event = JSON.parseObject(body, SseEvent.class);
+            if (event == null) {
+                log.warn("[SSE] 无法解析事件消息: {}", body);
+                return;
+            }
+
+            if (event.targetId() != null) {
+                connectionManager.sendTo(event.targetId(), event);
+            } else {
+                connectionManager.broadcast(event);
+            }
+        } catch (Exception e) {
+            log.error("[SSE] 处理 Redis 消息失败: {}", body, e);
+        }
+    }
+}