feat: 添加 RedisSseListener 和 SseConfig Redis Pub/Sub 配置

This commit is contained in:
2026-06-22 13:42:37 +08:00
parent 42ae5ef1af
commit 808dff53d2
2 changed files with 75 additions and 0 deletions

View File

@@ -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;
}
}

View File

@@ -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);
}
}
}