!348 增加webSocket连接测试以及文件优化

Merge pull request !348 from 咱哥丶/dev

!348 增加webSocket连接测试以及文件优化
Merge pull request !348 from 咱哥丶/dev
芋道源码 · Gitee
2 parents 0f722d0b 18622c5d
Showing 11 changed files with 433 additions and 190 deletions
sql/mysql/ruoyi-vue-pro.sql
yudao-dependencies/pom.xml
yudao-framework/yudao-spring-boot-starter-security/src/main/java/cn/iocoder/yudao/framework/security/config/YudaoWebSecurityConfigurerAdapter.java
yudao-module-infra/yudao-module-infra-biz/pom.xml
yudao-module-infra/yudao-module-infra-biz/src/main/java/cn/iocoder/yudao/module/infra/controller/admin/file/FileController.java
yudao-module-infra/yudao-module-infra-biz/src/main/java/cn/iocoder/yudao/module/infra/service/file/FileServiceImpl.java
yudao-module-infra/yudao-module-infra-biz/src/main/java/cn/iocoder/yudao/module/infra/websocket/SemaphoreUtils.java
yudao-module-infra/yudao-module-infra-biz/src/main/java/cn/iocoder/yudao/module/infra/websocket/WebSocketConfig.java
yudao-module-infra/yudao-module-infra-biz/src/main/java/cn/iocoder/yudao/module/infra/websocket/WebSocketServer.java
yudao-module-infra/yudao-module-infra-biz/src/main/java/cn/iocoder/yudao/module/infra/websocket/WebSocketUsers.java
yudao-ui-admin/src/views/infra/webSocket/index.vue
@@ -1710,6 +1710,8 @@ INSERT INTO `system_menu` (`id`, `name`, `permission`, `type`, `sort`, `parent_i
 INSERT INTO `system_menu` (`id`, `name`, `permission`, `type`, `sort`, `parent_id`, `path`, `icon`, `component`, `status`, `visible`, `keep_alive`, `creator`, `create_time`, `updater`, `update_time`, `deleted`) VALUES (1267, '客户端删除', 'system:oauth2-client:delete', 3, 4, 1263, '', '', '', 0, b'1', b'1', '', '2022-05-10 16:26:33', '1', '2022-05-11 00:31:33', b'0');
 INSERT INTO `system_menu` (`id`, `name`, `permission`, `type`, `sort`, `parent_id`, `path`, `icon`, `component`, `status`, `visible`, `keep_alive`, `creator`, `create_time`, `updater`, `update_time`, `deleted`) VALUES (1281, '可视化报表', '', 1, 12, 0, '/visualization', 'chart', NULL, 0, b'1', b'1', '1', '2022-07-10 20:22:15', '1', '2022-07-10 20:33:30', b'0');
 INSERT INTO `system_menu` (`id`, `name`, `permission`, `type`, `sort`, `parent_id`, `path`, `icon`, `component`, `status`, `visible`, `keep_alive`, `creator`, `create_time`, `updater`, `update_time`, `deleted`) VALUES (1282, '积木报表', '', 2, 1, 1281, 'jimu-report', 'example', 'visualization/jmreport/index', 0, b'1', b'1', '1', '2022-07-10 20:26:36', '1', '2022-07-28 21:17:34', b'0');
+INSERT INTO `system_menu` (`id`, `name`, `permission`, `type`, `sort`, `parent_id`, `path`, `icon`, `component`, `status`, `visible`, `keep_alive`, `creator`, `create_time`, `updater`, `update_time`, `deleted`) VALUES (1283, 'webSocket连接', '', 2, 14, 2, 'webSocket', '#', 'infra/webSocket/index', 0, b'1', b'1', '1', '2023-01-01 11:43:04', '1', '2023-01-01 11:43:04', b'0');
+
 COMMIT;
 -- ----------------------------
@@ -595,6 +595,12 @@
                 <artifactId>xercesImpl</artifactId>
                 <version>${xercesImpl.version}</version>
             </dependency>
+            <!-- SpringBoot Websocket -->
+            <dependency>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-starter-websocket</artifactId>
+                <version>${spring.boot.version}</version>
+            </dependency>
         </dependencies>
     </dependencyManagement>
@@ -129,6 +129,8 @@ public class YudaoWebSecurityConfigurerAdapter {
                 .antMatchers(buildAppApi("/**")).permitAll()
                 // 1.5 验证码captcha 允许匿名访问
                 .antMatchers("/captcha/get", "/captcha/check").permitAll()
+                // 1.6 webSocket 允许匿名访问
+                .antMatchers("/websocket/message").permitAll()
                 // ②：每个项目的自定义规则
                 .and().authorizeRequests(registry -> // 下面，循环设置自定义规则
                         authorizeRequestsCustomizers.forEach(customizer -> customizer.customize(registry)))
@@ -111,6 +111,12 @@
             <groupId>cn.iocoder.boot</groupId>
             <artifactId>yudao-spring-boot-starter-file</artifactId>
         </dependency>
+
+        <!-- WebSocket -->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-websocket</artifactId>
+        </dependency>
     </dependencies>
 </project>
-package cn.iocoder.yudao.module.infra.controller.admin.file;
-
-import cn.hutool.core.io.IoUtil;
-import cn.hutool.core.util.StrUtil;
-import cn.iocoder.yudao.framework.common.pojo.CommonResult;
-import cn.iocoder.yudao.framework.common.pojo.PageResult;
-import cn.iocoder.yudao.framework.common.util.servlet.ServletUtils;
-import cn.iocoder.yudao.framework.operatelog.core.annotations.OperateLog;
-import cn.iocoder.yudao.module.infra.controller.admin.file.vo.file.FilePageReqVO;
-import cn.iocoder.yudao.module.infra.controller.admin.file.vo.file.FileRespVO;
-import cn.iocoder.yudao.module.infra.controller.admin.file.vo.file.FileUploadReqVO;
-import cn.iocoder.yudao.module.infra.convert.file.FileConvert;
-import cn.iocoder.yudao.module.infra.dal.dataobject.file.FileDO;
-import cn.iocoder.yudao.module.infra.service.file.FileService;
-import io.swagger.annotations.Api;
-import io.swagger.annotations.ApiImplicitParam;
-import io.swagger.annotations.ApiOperation;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.http.HttpStatus;
-import org.springframework.security.access.prepost.PreAuthorize;
-import org.springframework.validation.annotation.Validated;
-import org.springframework.web.bind.annotation.*;
-import org.springframework.web.multipart.MultipartFile;
-
-import javax.annotation.Resource;
-import javax.annotation.security.PermitAll;
-import javax.servlet.http.HttpServletRequest;
-import javax.servlet.http.HttpServletResponse;
-import javax.validation.Valid;
-
-import static cn.iocoder.yudao.framework.common.pojo.CommonResult.success;
-
-@Api(tags = "管理后台 - 文件存储")
-@RestController
-@RequestMapping("/infra/file")
-@Validated
-@Slf4j
-public class FileController {
-
-    @Resource
-    private FileService fileService;
-
-    @PostMapping("/upload")
-    @ApiOperation("上传文件")
-    @OperateLog(logArgs = false) // 上传文件，没有记录操作日志的必要
-    public CommonResult<String> uploadFile(FileUploadReqVO uploadReqVO) throws Exception {
-        MultipartFile file = uploadReqVO.getFile();
-        String path = uploadReqVO.getPath();
-        return success(fileService.createFile(file.getOriginalFilename(), path, IoUtil.readBytes(file.getInputStream())));
-    }
-
-    @DeleteMapping("/delete")
-    @ApiOperation("删除文件")
-    @ApiImplicitParam(name = "id", value = "编号", required = true, dataTypeClass = Long.class)
-    @PreAuthorize("@ss.hasPermission('infra:file:delete')")
-    public CommonResult<Boolean> deleteFile(@RequestParam("id") Long id) throws Exception {
-        fileService.deleteFile(id);
-        return success(true);
-    }
-
-    @GetMapping("/{configId}/get/**")
-    @PermitAll
-    @ApiOperation("下载文件")
-    @ApiImplicitParam(name = "configId", value = "配置编号",  required = true, dataTypeClass = Long.class)
-    public void getFileContent(HttpServletRequest request,
-                               HttpServletResponse response,
-                               @PathVariable("configId") Long configId) throws Exception {
-        // 获取请求的路径
-        String path = StrUtil.subAfter(request.getRequestURI(), "/get/", false);
-        if (StrUtil.isEmpty(path)) {
-            throw new IllegalArgumentException("结尾的 path 路径必须传递");
-        }
-
-        // 读取内容
-        byte[] content = fileService.getFileContent(configId, path);
-        if (content == null) {
-            log.warn("[getFileContent][configId({}) path({}) 文件不存在]", configId, path);
-            response.setStatus(HttpStatus.NOT_FOUND.value());
-            return;
-        }
-        ServletUtils.writeAttachment(response, path, content);
-    }
-
-    @GetMapping("/page")
-    @ApiOperation("获得文件分页")
-    @PreAuthorize("@ss.hasPermission('infra:file:query')")
-    public CommonResult<PageResult<FileRespVO>> getFilePage(@Valid FilePageReqVO pageVO) {
-        PageResult<FileDO> pageResult = fileService.getFilePage(pageVO);
-        return success(FileConvert.INSTANCE.convertPage(pageResult));
-    }
-
-}
-package cn.iocoder.yudao.module.infra.service.file;
-
-import cn.hutool.core.lang.Assert;
-import cn.hutool.core.util.StrUtil;
-import cn.iocoder.yudao.framework.common.pojo.PageResult;
-import cn.iocoder.yudao.framework.common.util.io.FileUtils;
-import cn.iocoder.yudao.framework.file.core.client.FileClient;
-import cn.iocoder.yudao.framework.file.core.utils.FileTypeUtils;
-import cn.iocoder.yudao.module.infra.controller.admin.file.vo.file.FilePageReqVO;
-import cn.iocoder.yudao.module.infra.dal.dataobject.file.FileDO;
-import cn.iocoder.yudao.module.infra.dal.mysql.file.FileMapper;
-import lombok.SneakyThrows;
-import org.springframework.stereotype.Service;
-
-import javax.annotation.Resource;
-
-import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
-import static cn.iocoder.yudao.module.infra.enums.ErrorCodeConstants.FILE_NOT_EXISTS;
-
-/**
- * 文件 Service 实现类
- *
- * @author 芋道源码
- */
-@Service
-public class FileServiceImpl implements FileService {
-
-    @Resource
-    private FileConfigService fileConfigService;
-
-    @Resource
-    private FileMapper fileMapper;
-
-    @Override
-    public PageResult<FileDO> getFilePage(FilePageReqVO pageReqVO) {
-        return fileMapper.selectPage(pageReqVO);
-    }
-
-    @Override
-    @SneakyThrows
-    public String createFile(String name, String path, byte[] content) {
-        // 计算默认的 path 名
-        String type = FileTypeUtils.getMineType(content, name);
-        if (StrUtil.isEmpty(path)) {
-            path = FileUtils.generatePath(content, name);
-        }
-        // 如果 name 为空，则使用 path 填充
-        if (StrUtil.isEmpty(name)) {
-            name = path;
-        }
-
-        // 上传到文件存储器
-        FileClient client = fileConfigService.getMasterFileClient();
-        Assert.notNull(client, "客户端(master) 不能为空");
-        String url = client.upload(content, path, type);
-
-        // 保存到数据库
-        FileDO file = new FileDO();
-        file.setConfigId(client.getId());
-        file.setName(name);
-        file.setPath(path);
-        file.setUrl(url);
-        file.setType(type);
-        file.setSize(content.length);
-        fileMapper.insert(file);
-        return url;
-    }
-
-    @Override
-    public void deleteFile(Long id) throws Exception {
-        // 校验存在
-        FileDO file = this.validateFileExists(id);
-
-        // 从文件存储器中删除
-        FileClient client = fileConfigService.getFileClient(file.getConfigId());
-        Assert.notNull(client, "客户端({}) 不能为空", file.getConfigId());
-        client.delete(file.getPath());
-
-        // 删除记录
-        fileMapper.deleteById(id);
-    }
-
-    private FileDO validateFileExists(Long id) {
-        FileDO fileDO = fileMapper.selectById(id);
-        if (fileDO == null) {
-            throw exception(FILE_NOT_EXISTS);
-        }
-        return fileDO;
-    }
-
-    @Override
-    public byte[] getFileContent(Long configId, String path) throws Exception {
-        FileClient client = fileConfigService.getFileClient(configId);
-        Assert.notNull(client, "客户端({}) 不能为空", configId);
-        return client.getContent(path);
-    }
-
-}
+package cn.iocoder.yudao.module.infra.websocket;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.concurrent.Semaphore;
+
+/**
+ * 信号量相关处理
+ *
+ */
+@Slf4j
+public class SemaphoreUtils {
+
+    /**
+     * 获取信号量
+     *
+     * @param semaphore
+     * @return
+     */
+    public static boolean tryAcquire(Semaphore semaphore) {
+        boolean flag = false;
+
+        try {
+            flag = semaphore.tryAcquire();
+        } catch (Exception e) {
+            log.error("获取信号量异常", e);
+        }
+
+        return flag;
+    }
+
+    /**
+     * 释放信号量
+     *
+     * @param semaphore
+     */
+    public static void release(Semaphore semaphore) {
+
+        try {
+            semaphore.release();
+        } catch (Exception e) {
+            log.error("释放信号量异常", e);
+        }
+    }
+}
+package cn.iocoder.yudao.module.infra.websocket;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.socket.server.standard.ServerEndpointExporter;
+
+/**
+ * websocket 配置
+ */
+@Configuration
+public class WebSocketConfig {
+    @Bean
+    public ServerEndpointExporter serverEndpointExporter() {
+        return new ServerEndpointExporter();
+    }
+}
+package cn.iocoder.yudao.module.infra.websocket;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import javax.websocket.*;
+import javax.websocket.server.ServerEndpoint;
+import java.util.concurrent.Semaphore;
+
+/**
+ * websocket 消息处理
+ */
+@Component
+@ServerEndpoint("/websocket/message")
+@Slf4j
+public class WebSocketServer {
+
+    /**
+     * 默认最多允许同时在线用户数100
+     */
+    public static int socketMaxOnlineCount = 100;
+
+    private static final Semaphore SOCKET_SEMAPHORE = new Semaphore(socketMaxOnlineCount);
+
+    /**
+     * 连接建立成功调用的方法
+     */
+    @OnOpen
+    public void onOpen(Session session) throws Exception {
+        // 尝试获取信号量
+        boolean semaphoreFlag = SemaphoreUtils.tryAcquire(SOCKET_SEMAPHORE);
+        if (!semaphoreFlag) {
+            // 未获取到信号量
+            log.error("当前在线人数超过限制数：{}", socketMaxOnlineCount);
+            WebSocketUsers.sendMessage(session, "当前在线人数超过限制数：" + socketMaxOnlineCount);
+            session.close();
+        } else {
+            String userId = WebSocketUsers.getParam("userId", session);
+            if (userId != null) {
+                // 添加用户
+                WebSocketUsers.addSession(userId, session);
+                log.info("用户【userId={}】建立连接，当前连接用户总数：{}", userId, WebSocketUsers.getUsers().size());
+                WebSocketUsers.sendMessage(session, "接收内容：连接成功");
+            } else {
+                WebSocketUsers.sendMessage(session, "接收内容：连接失败");
+            }
+        }
+    }
+
+    /**
+     * 连接关闭时处理
+     */
+    @OnClose
+    public void onClose(Session session) {
+        log.info("用户【sessionId={}】关闭连接！", session.getId());
+        // 移除用户
+        WebSocketUsers.removeSession(session);
+        // 获取到信号量则需释放
+        SemaphoreUtils.release(SOCKET_SEMAPHORE);
+    }
+
+    /**
+     * 抛出异常时处理
+     */
+    @OnError
+    public void onError(Session session, Throwable exception) throws Exception {
+        if (session.isOpen()) {
+            // 关闭连接
+            session.close();
+        }
+        String sessionId = session.getId();
+        log.info("用户【sessionId={}】连接异常！异常信息：{}", sessionId, exception);
+        // 移出用户
+        WebSocketUsers.removeSession(session);
+        // 获取到信号量则需释放
+        SemaphoreUtils.release(SOCKET_SEMAPHORE);
+    }
+
+    /**
+     * 收到客户端消息时调用的方法
+     */
+    @OnMessage
+    public void onMessage(Session session, String message) {
+        WebSocketUsers.sendMessage(session, "接收内容：" + message);
+    }
+}
+package cn.iocoder.yudao.module.infra.websocket;
+
+import cn.hutool.core.map.MapUtil;
+import cn.hutool.core.util.StrUtil;
+import lombok.extern.slf4j.Slf4j;
+import org.bouncycastle.util.Strings;
+
+import javax.validation.constraints.NotNull;
+import javax.websocket.Session;
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * websocket 客户端用户
+ */
+@Slf4j
+public class WebSocketUsers {
+
+    /**
+     * 用户集
+     *  TODO 需要登录用户的session？
+     */
+    private static final Map<String, Session> SESSION_MAP = new ConcurrentHashMap<>();
+
+    /**
+     * 存储用户
+     *
+     * @param userId  唯一键
+     * @param session 用户信息
+     */
+    public static void addSession(String userId, Session session) {
+        SESSION_MAP.put(userId, session);
+    }
+
+    /**
+     * 移除用户
+     *
+     * @param session 用户信息
+     * @return 移除结果
+     */
+    public static boolean removeSession(Session session) {
+        String key = null;
+        boolean flag = SESSION_MAP.containsValue(session);
+        if (flag) {
+            Set<Map.Entry<String, Session>> entries = SESSION_MAP.entrySet();
+            for (Map.Entry<String, Session> entry : entries) {
+                Session value = entry.getValue();
+                if (value.equals(session)) {
+                    key = entry.getKey();
+                    break;
+                }
+            }
+        } else {
+            return true;
+        }
+        return removeSession(key);
+    }
+
+    /**
+     * 移出用户
+     *
+     * @param userId 用户id
+     */
+    public static boolean removeSession(String userId) {
+        log.info("用户【userId={}】退出", userId);
+        Session remove = SESSION_MAP.remove(userId);
+        if (remove != null) {
+            boolean containsValue = SESSION_MAP.containsValue(remove);
+            log.info("用户【userId={}】退出{}，当前连接用户总数：{}", userId, containsValue ? "失败" : "成功", SESSION_MAP.size());
+            return containsValue;
+        } else {
+            return true;
+        }
+    }
+
+    /**
+     * 获取在线用户列表
+     *
+     * @return 返回用户集合
+     */
+    public static Map<String, Session> getUsers() {
+        return SESSION_MAP;
+    }
+
+    /**
+     * 向所有在线人发送消息
+     *
+     * @param message 消息内容
+     */
+    public static void sendMessageToAll(String message) {
+        SESSION_MAP.forEach((userId, session) -> {
+            if (session.isOpen()) {
+                sendMessage(session, message);
+            }
+        });
+    }
+
+    /**
+     * 异步发送文本消息
+     *
+     * @param session 用户session
+     * @param message 消息内容
+     */
+    public static void sendMessageAsync(Session session, String message) {
+        if (session.isOpen()) {
+            // TODO 需要加synchronized锁（synchronized(session)）？单个session创建线程？
+            session.getAsyncRemote().sendText(message);
+        } else {
+            log.warn("用户【session={}】不在线", session.getId());
+        }
+    }
+
+    /**
+     * 同步发送文本消息
+     *
+     * @param session 用户session
+     * @param message 消息内容
+     */
+    public static void sendMessage(Session session, String message) {
+        try {
+            if (session.isOpen()) {
+                // TODO 需要加synchronized锁（synchronized(session)）？单个session创建线程？
+                session.getBasicRemote().sendText(message);
+            } else {
+                log.warn("用户【session={}】不在线", session.getId());
+            }
+        } catch (IOException e) {
+            log.error("发送消息异常", e);
+        }
+
+    }
+
+    /**
+     * 根据用户id发送消息
+     *
+     * @param userId  用户id
+     * @param message 消息内容
+     */
+    public static void sendMessage(String userId, String message) {
+        Session session = SESSION_MAP.get(userId);
+        //判断是否存在该用户的session，并且是否在线
+        if (session == null || !session.isOpen()) {
+            return;
+        }
+        sendMessage(session, message);
+    }
+
+
+    /**
+     * 获取session中的指定参数值
+     *
+     * @param key     参数key
+     * @param session 用户session
+     */
+    public static String getParam(@NotNull String key, Session session) {
+        //TODO 目前只针对获取一个key的值，后期根据情况拓展多个 或者直接在onClose onOpen上获取参数？
+        String value = null;
+        Map<String, List<String>> parameters = session.getRequestParameterMap();
+        if (MapUtil.isNotEmpty(parameters)) {
+            value = parameters.get(key).get(0);
+        } else {
+            String queryString = session.getQueryString();
+            if (!StrUtil.isEmpty(queryString)) {
+                String[] params = Strings.split(queryString, '&');
+                for (String paramPair : params) {
+                    String[] nameValues = Strings.split(paramPair, '=');
+                    if (key.equals(nameValues[0])) {
+                        value = nameValues[1];
+                    }
+                }
+            }
+        }
+        return value;
+    }
+}
+<template>
+  <div class="app-container">
+    <el-form label-width="120px">
+      <el-row type="flex" :gutter="0">
+        <el-col :sm="12">
+          <el-form-item label="WebSocket地址" size="small">
+            <el-input v-model="url" type="text"/>
+          </el-form-item>
+        </el-col>
+        <el-col :offset="1">
+          <el-form-item label="" label-width="0px" size="small">
+            <el-button @click="connect" type="primary" :disabled="ws&&ws.readyState===1">
+              {{ ws && ws.readyState === 1 ? "已连接" : "连接" }}
+            </el-button>
+            <el-button @click="exit" type="danger">断开</el-button>
+          </el-form-item>
+        </el-col>
+      </el-row>
+      <el-form-item label="发送内容" size="small">
+        <el-input type="textarea" v-model="message" :rows="5"/>
+      </el-form-item>
+      <el-form-item label="" size="small">
+        <el-button type="success" @click="send">发送消息</el-button>
+      </el-form-item>
+      <el-form-item label="接收内容" size="small">
+        <el-input type="textarea" v-model="content" :rows="12" disabled/>
+      </el-form-item>
+      <el-form-item label="" size="small">
+        <el-button type="info" @click="content=''">清空消息</el-button>
+      </el-form-item>
+    </el-form>
+  </div>
+</template>
+
+<script>
+import store from "@/store";
+import {getNowDateTime} from "@/utils/ruoyi";
+
+export default {
+  data() {
+    return {
+      url: process.env.VUE_APP_BASE_API + "/websocket/message",
+      message: "",
+      content: "",
+      ws: null,
+    };
+  },
+  created() {
+    this.url = this.url.replace("http", "ws")
+  },
+  methods: {
+    connect() {
+      if (!'WebSocket' in window) {
+        this.$modal.msgError("您的浏览器不支持WebSocket");
+        return;
+      }
+      const userId = store.getters.userId;
+      this.ws = new WebSocket(this.url + "?userId=" + userId);
+      const self = this;
+      this.ws.onopen = function (event) {
+        self.content = self.content + "\n**********************连接开始**********************\n";
+      };
+      this.ws.onmessage = function (event) {
+        self.content = self.content + "接收时间：" + getNowDateTime() + "\n" + event.data + "\n";
+      };
+      this.ws.onclose = function (event) {
+        self.content = self.content + "**********************连接关闭**********************\n";
+      };
+      this.ws.error = function (event) {
+        self.content = self.content + "**********************连接异常**********************\n";
+      };
+    },
+    exit() {
+      if (this.ws) {
+        this.ws.close();
+        this.ws = null;
+      }
+    },
+    send() {
+      if (!this.ws || this.ws.readyState !== 1) {
+        this.$modal.msgError("未连接到服务器");
+        return;
+      }
+      if (!this.message) {
+        this.$modal.msgError("请输入发送内容");
+        return;
+      }
+      this.ws.send(this.message);
+    }
+  },
+};
+</script>