add 增加 ruoyi-common-sse 模块 支持SSE推送 比ws更轻量更稳定的推送

ruoyi-common/pom.xml

@@ -33,6 +33,7 @@
         <module>ruoyi-common-encrypt</module>
         <module>ruoyi-common-tenant</module>
         <module>ruoyi-common-websocket</module>
+        <module>ruoyi-common-sse</module>
     </modules>
 
     <artifactId>ruoyi-common</artifactId>

ruoyi-common/ruoyi-common-bom/pom.xml

@@ -172,6 +172,13 @@
                 <version>${revision}</version>
             </dependency>
 
+            <!-- SSE模块 -->
+            <dependency>
+                <groupId>org.dromara</groupId>
+                <artifactId>ruoyi-common-sse</artifactId>
+                <version>${revision}</version>
+            </dependency>
+
         </dependencies>
     </dependencyManagement>
 

ruoyi-common/ruoyi-common-sse/pom.xml

@@ -0,0 +1,36 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <groupId>org.dromara</groupId>
+        <artifactId>ruoyi-common</artifactId>
+        <version>${revision}</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>ruoyi-common-sse</artifactId>
+
+    <description>
+        ruoyi-common-sse 模块
+    </description>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.dromara</groupId>
+            <artifactId>ruoyi-common-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.dromara</groupId>
+            <artifactId>ruoyi-common-redis</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.dromara</groupId>
+            <artifactId>ruoyi-common-satoken</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.dromara</groupId>
+            <artifactId>ruoyi-common-json</artifactId>
+        </dependency>
+    </dependencies>
+</project>

ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/config/SseAutoConfiguration.java

@@ -0,0 +1,28 @@
+package org.dromara.common.sse.config;
+
+import org.dromara.common.sse.core.SseEmitterManager;
+import org.dromara.common.sse.listener.SseTopicListener;
+import org.springframework.boot.autoconfigure.AutoConfiguration;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+
+/**
+ * @author Lion Li
+ */
+@AutoConfiguration
+@ConditionalOnProperty(value = "sse.enabled", havingValue = "true")
+@EnableConfigurationProperties(SseProperties.class)
+public class SseAutoConfiguration {
+
+    @Bean
+    public SseEmitterManager sseEmitterManager() {
+        return new SseEmitterManager();
+    }
+
+    @Bean
+    public SseTopicListener sseTopicListener() {
+        return new SseTopicListener();
+    }
+
+}

ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/config/SseProperties.java

@@ -0,0 +1,21 @@
+package org.dromara.common.sse.config;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * SSE 配置项
+ *
+ * @author Lion Li
+ */
+@Data
+@ConfigurationProperties("sse")
+public class SseProperties {
+
+    private Boolean enabled;
+
+    /**
+     * 路径
+     */
+    private String path;
+}

ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/controller/SseController.java

@@ -0,0 +1,52 @@
+package org.dromara.common.sse.controller;
+
+import cn.dev33.satoken.stp.StpUtil;
+import lombok.RequiredArgsConstructor;
+import org.dromara.common.core.domain.R;
+import org.dromara.common.satoken.utils.LoginHelper;
+import org.dromara.common.sse.core.SseEmitterManager;
+import org.dromara.common.sse.dto.SseMessageDto;
+import org.springframework.http.MediaType;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+import java.util.List;
+
+@RestController
+@RequiredArgsConstructor
+public class SseController {
+
+    private final SseEmitterManager sseEmitterManager;
+
+    @GetMapping(value = "${sse.path}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
+    public SseEmitter connect() {
+        String tokenValue = StpUtil.getTokenValue();
+        Long userId = LoginHelper.getUserId();
+        return sseEmitterManager.connect(userId, tokenValue);
+    }
+
+    @GetMapping(value = "${sse.path}/close")
+    public R<Void> close() {
+        String tokenValue = StpUtil.getTokenValue();
+        Long userId = LoginHelper.getUserId();
+        sseEmitterManager.disconnect(userId, tokenValue);
+        return R.ok();
+    }
+
+    @GetMapping(value = "${sse.path}/send")
+    public R<Void> send(Long userId, String msg) {
+        SseMessageDto dto = new SseMessageDto();
+        dto.setUserIds(List.of(userId));
+        dto.setMessage(msg);
+        sseEmitterManager.publishMessage(dto);
+        return R.ok();
+    }
+
+    @GetMapping(value = "${sse.path}/sendAll")
+    public R<Void> send(String msg) {
+        sseEmitterManager.publishAll(msg);
+        return R.ok();
+    }
+
+}

ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/core/SseEmitterManager.java

@@ -0,0 +1,134 @@
+package org.dromara.common.sse.core;
+
+import cn.hutool.core.collection.CollUtil;
+import lombok.extern.slf4j.Slf4j;
+import org.dromara.common.redis.utils.RedisUtils;
+import org.dromara.common.sse.dto.SseMessageDto;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Consumer;
+
+@Slf4j
+public class SseEmitterManager {
+    /**
+     * 订阅的频道
+     */
+    private final static String SSE_TOPIC = "global:sse";
+
+    private final static Map<Long, Map<String, SseEmitter>> USER_TOKEN_EMITTERS = new ConcurrentHashMap<>();
+
+    public SseEmitter connect(Long userId, String token) {
+        Map<String, SseEmitter> emitters = USER_TOKEN_EMITTERS.computeIfAbsent(userId, k -> new ConcurrentHashMap<>());
+        SseEmitter emitter = new SseEmitter(0L);
+
+        emitters.put(token, emitter);
+
+        emitter.onCompletion(() -> emitters.remove(token));
+        emitter.onTimeout(() -> emitters.remove(token));
+
+        try {
+            emitter.send(SseEmitter.event().comment("connected"));
+        } catch (IOException e) {
+            emitters.remove(token);
+        }
+        return emitter;
+    }
+
+    public void disconnect(Long userId, String token) {
+        Map<String, SseEmitter> emitters = USER_TOKEN_EMITTERS.get(userId);
+        if (emitters != null) {
+            try {
+                emitters.get(token).send(SseEmitter.event().comment("disconnected"));
+            } catch (IOException ignore) {
+            }
+            emitters.remove(token);
+        }
+    }
+
+    /**
+     * 订阅SSE消息主题，并提供一个消费者函数来处理接收到的消息
+     *
+     * @param consumer 处理SSE消息的消费者函数
+     */
+    public void subscribeMessage(Consumer<SseMessageDto> consumer) {
+        RedisUtils.subscribe(SSE_TOPIC, SseMessageDto.class, consumer);
+    }
+
+    /**
+     * 向指定的用户会话发送消息
+     *
+     * @param userId  要发送消息的用户id
+     * @param message 要发送的消息内容
+     */
+    public void sendMessage(Long userId, String message) {
+        Map<String, SseEmitter> emitters = USER_TOKEN_EMITTERS.get(userId);
+        if (emitters != null) {
+            for (Map.Entry<String, SseEmitter> entry : emitters.entrySet()) {
+                try {
+                    entry.getValue().send(SseEmitter.event()
+                        .name("message")
+                        .reconnectTime(10000L)
+                        .data(message));
+                } catch (Exception e) {
+                    emitters.remove(entry.getKey());
+                }
+            }
+        }
+    }
+
+    /**
+     * 本机全用户会话发送消息
+     *
+     * @param message 要发送的消息内容
+     */
+    public void sendMessage(String message) {
+        for (Long userId : USER_TOKEN_EMITTERS.keySet()) {
+            sendMessage(userId, message);
+        }
+    }
+
+    /**
+     * 发布SSE订阅消息
+     *
+     * @param sseMessageDto 要发布的SSE消息对象
+     */
+    public void publishMessage(SseMessageDto sseMessageDto) {
+        List<Long> unsentUserIds = new ArrayList<>();
+        // 当前服务内用户,直接发送消息
+        for (Long userId : sseMessageDto.getUserIds()) {
+            if (USER_TOKEN_EMITTERS.containsKey(userId)) {
+                sendMessage(userId, sseMessageDto.getMessage());
+                continue;
+            }
+            unsentUserIds.add(userId);
+        }
+        // 不在当前服务内用户,发布订阅消息
+        if (CollUtil.isNotEmpty(unsentUserIds)) {
+            SseMessageDto broadcastMessage = new SseMessageDto();
+            broadcastMessage.setMessage(sseMessageDto.getMessage());
+            broadcastMessage.setUserIds(unsentUserIds);
+            RedisUtils.publish(SSE_TOPIC, broadcastMessage, consumer -> {
+                log.info("SSE发送主题订阅消息topic:{} session keys:{} message:{}",
+                    SSE_TOPIC, unsentUserIds, sseMessageDto.getMessage());
+            });
+        }
+    }
+
+    /**
+     * 向所有的用户发布订阅的消息(群发)
+     *
+     * @param message 要发布的消息内容
+     */
+    public void publishAll(String message) {
+        SseMessageDto broadcastMessage = new SseMessageDto();
+        broadcastMessage.setMessage(message);
+        RedisUtils.publish(SSE_TOPIC, broadcastMessage, consumer -> {
+            log.info("SSE发送主题订阅消息topic:{} message:{}", SSE_TOPIC, message);
+        });
+    }
+}

ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/dto/SseMessageDto.java

@@ -0,0 +1,29 @@
+package org.dromara.common.sse.dto;
+
+import lombok.Data;
+
+import java.io.Serial;
+import java.io.Serializable;
+import java.util.List;
+
+/**
+ * 消息的dto
+ *
+ * @author zendwang
+ */
+@Data
+public class SseMessageDto implements Serializable {
+
+    @Serial
+    private static final long serialVersionUID = 1L;
+
+    /**
+     * 需要推送到的session key 列表
+     */
+    private List<Long> userIds;
+
+    /**
+     * 需要发送的消息
+     */
+    private String message;
+}

ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/listener/SseTopicListener.java

@@ -0,0 +1,48 @@
+package org.dromara.common.sse.listener;
+
+import cn.hutool.core.collection.CollUtil;
+import lombok.extern.slf4j.Slf4j;
+import org.dromara.common.sse.core.SseEmitterManager;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.core.Ordered;
+
+/**
+ * SSE 主题订阅监听器
+ *
+ * @author Lion Li
+ */
+@Slf4j
+public class SseTopicListener implements ApplicationRunner, Ordered {
+
+    @Autowired
+    private SseEmitterManager sseEmitterManager;
+
+    /**
+     * 在Spring Boot应用程序启动时初始化SSE主题订阅监听器
+     *
+     * @param args 应用程序参数
+     * @throws Exception 初始化过程中可能抛出的异常
+     */
+    @Override
+    public void run(ApplicationArguments args) throws Exception {
+        sseEmitterManager.subscribeMessage((message) -> {
+            log.info("SSE主题订阅收到消息session keys={} message={}", message.getUserIds(), message.getMessage());
+            // 如果key不为空就按照key发消息 如果为空就群发
+            if (CollUtil.isNotEmpty(message.getUserIds())) {
+                message.getUserIds().forEach(key -> {
+                    sseEmitterManager.sendMessage(key, message.getMessage());
+                });
+            } else {
+                sseEmitterManager.sendMessage(message.getMessage());
+            }
+        });
+        log.info("初始化SSE主题订阅监听器成功");
+    }
+
+    @Override
+    public int getOrder() {
+        return -1;
+    }
+}

ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/utils/SseMessageUtils.java

@@ -0,0 +1,58 @@
+package org.dromara.common.sse.utils;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.dromara.common.core.utils.SpringUtils;
+import org.dromara.common.sse.core.SseEmitterManager;
+import org.dromara.common.sse.dto.SseMessageDto;
+
+/**
+ * 工具类
+ *
+ * @author Lion Li
+ */
+@Slf4j
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class SseMessageUtils {
+
+    private final static SseEmitterManager MANAGER = SpringUtils.getBean(SseEmitterManager.class);
+
+    /**
+     * 向指定的WebSocket会话发送消息
+     *
+     * @param userId  要发送消息的用户id
+     * @param message 要发送的消息内容
+     */
+    public static void sendMessage(Long userId, String message) {
+        MANAGER.sendMessage(userId, message);
+    }
+
+    /**
+     * 本机全用户会话发送消息
+     *
+     * @param message 要发送的消息内容
+     */
+    public static void sendMessage(String message) {
+        MANAGER.sendMessage(message);
+    }
+
+    /**
+     * 发布SSE订阅消息
+     *
+     * @param sseMessageDto 要发布的SSE消息对象
+     */
+    public static void publishMessage(SseMessageDto sseMessageDto) {
+        MANAGER.publishMessage(sseMessageDto);
+    }
+
+    /**
+     * 向所有的用户发布订阅的消息(群发)
+     *
+     * @param message 要发布的消息内容
+     */
+    public static void publishAll(String message) {
+        MANAGER.publishAll(message);
+    }
+
+}

ruoyi-common/ruoyi-common-sse/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -0,0 +1 @@
+org.dromara.common.sse.config.SseAutoConfiguration

ruoyi-modules/ruoyi-system/pom.xml

@@ -95,6 +95,11 @@
             <artifactId>ruoyi-common-websocket</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.dromara</groupId>
+            <artifactId>ruoyi-common-sse</artifactId>
+        </dependency>
+
     </dependencies>
 
 </project>