消息推送功能

2023-03-08 11:12:56 +08:00
parent 2ae38c75a3
commit 55d5be8b79
20 changed files with 1345 additions and 20 deletions
--- a/ktg-framework/pom.xml
+++ b/ktg-framework/pom.xml
@@ -47,6 +47,12 @@
            </exclusions>
        </dependency>

+        <!-- WebSocket -->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-websocket</artifactId>
+        </dependency>
+
        <!-- 获取系统信息 -->
        <dependency>
            <groupId>com.github.oshi</groupId>
--- a/ktg-framework/src/main/java/com/ktg/framework/config/SecurityConfig.java
+++ b/ktg-framework/src/main/java/com/ktg/framework/config/SecurityConfig.java
@@ -114,6 +114,7 @@ public class SecurityConfig extends WebSecurityConfigurerAdapter
                .antMatchers("/webjars/**").anonymous()
                .antMatchers("/*/api-docs").anonymous()
                .antMatchers("/druid/**").anonymous()
+                .antMatchers("/websocket/**").anonymous()
                .antMatchers("/system/autocode/get/**").permitAll()
                // 除上面外的所有请求全部需要鉴权认证
                .anyRequest().authenticated()
--- a/ktg-framework/src/main/java/com/ktg/framework/message/MessageProvider.java
+++ b/ktg-framework/src/main/java/com/ktg/framework/message/MessageProvider.java
@@ -0,0 +1,41 @@
+package com.ktg.framework.message;
+
+import com.alibaba.fastjson.JSON;
+import com.ktg.common.constant.UserConstants;
+import com.ktg.common.core.domain.entity.SysUser;
+import com.ktg.common.utils.StringUtils;
+import com.ktg.framework.websocket.WebSocketUsers;
+import com.ktg.system.domain.SysMessage;
+import com.ktg.system.service.ISysMessageService;
+import com.ktg.system.service.ISysUserService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
+
+@Component
+public class MessageProvider {
+
+    @Autowired
+    private ISysMessageService sysMessageService;
+    @Autowired
+    private ISysUserService sysUserService;
+
+    @Transactional
+    public void sendMessage(SysMessage message){
+        message.setStatus(UserConstants.MESSAGE_STATUS_UNREAD);
+        if(StringUtils.isNotNull(message.getRecipientId())){
+            SysUser recipient = sysUserService.selectUserById(message.getRecipientId());
+            message.setRecipientName(recipient.getUserName());
+            message.setRecipientNick(recipient.getNickName());
+        }
+        if(StringUtils.isNotNull(message.getSenderId())){
+            SysUser sender = sysUserService.selectUserById(message.getSenderId());
+            message.setRecipientName(sender.getUserName());
+            message.setRecipientNick(sender.getNickName());
+        }
+        message.setDeletedFlag(UserConstants.NO);
+        WebSocketUsers.sendMesssageToUserByName(message.getRecipientName(), JSON.toJSONString(message));
+        sysMessageService.insertSysMessage(message);
+    }
+
+}
--- a/ktg-framework/src/main/java/com/ktg/framework/web/service/TokenService.java
+++ b/ktg-framework/src/main/java/com/ktg/framework/web/service/TokenService.java
@@ -222,4 +222,28 @@ public class TokenService
    {
        return Constants.LOGIN_TOKEN_KEY + uuid;
    }
+
+    /**
+     * 根据Token获取对应的用户
+     * @param token
+     * @return
+     */
+    public LoginUser getUserByToken(String token){
+        if (StringUtils.isNotEmpty(token))
+        {
+            try
+            {
+                Claims claims = parseToken(token);
+                // 解析对应的权限以及用户信息
+                String uuid = (String) claims.get(Constants.LOGIN_USER_KEY);
+                String userKey = getTokenKey(uuid);
+                LoginUser user = redisCache.getCacheObject(userKey);
+                return user;
+            }
+            catch (Exception e)
+            {
+            }
+        }
+        return null;
+    }
 }
--- a/ktg-framework/src/main/java/com/ktg/framework/websocket/SemaphoreUtils.java
+++ b/ktg-framework/src/main/java/com/ktg/framework/websocket/SemaphoreUtils.java
@@ -0,0 +1,58 @@
+package com.ktg.framework.websocket;
+
+import java.util.concurrent.Semaphore;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * 信号量相关处理
+ * 
+ * @author ruoyi
+ */
+public class SemaphoreUtils
+{
+    /**
+     * SemaphoreUtils 日志控制器
+     */
+    private static final Logger LOGGER = LoggerFactory.getLogger(SemaphoreUtils.class);
+
+    /**
+     * 获取信号量
+     * 
+     * @param semaphore
+     * @return
+     */
+    public static boolean tryAcquire(Semaphore semaphore)
+    {
+        boolean flag = false;
+
+        try
+        {
+            flag = semaphore.tryAcquire();
+        }
+        catch (Exception e)
+        {
+            LOGGER.error("获取信号量异常", e);
+        }
+
+        return flag;
+    }
+
+    /**
+     * 释放信号量
+     * 
+     * @param semaphore
+     */
+    public static void release(Semaphore semaphore)
+    {
+
+        try
+        {
+            semaphore.release();
+        }
+        catch (Exception e)
+        {
+            LOGGER.error("释放信号量异常", e);
+        }
+    }
+}
--- a/ktg-framework/src/main/java/com/ktg/framework/websocket/WebSocketConfig.java
+++ b/ktg-framework/src/main/java/com/ktg/framework/websocket/WebSocketConfig.java
@@ -0,0 +1,20 @@
+package com.ktg.framework.websocket;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.socket.server.standard.ServerEndpointExporter;
+
+/**
+ * websocket 配置
+ * 
+ * @author ruoyi
+ */
+@Configuration
+public class WebSocketConfig
+{
+    @Bean
+    public ServerEndpointExporter serverEndpointExporter()
+    {
+        return new ServerEndpointExporter();
+    }
+}
--- a/ktg-framework/src/main/java/com/ktg/framework/websocket/WebSocketServer.java
+++ b/ktg-framework/src/main/java/com/ktg/framework/websocket/WebSocketServer.java
@@ -0,0 +1,138 @@
+package com.ktg.framework.websocket;
+
+import java.util.concurrent.Semaphore;
+import javax.websocket.OnClose;
+import javax.websocket.OnError;
+import javax.websocket.OnMessage;
+import javax.websocket.OnOpen;
+import javax.websocket.Session;
+import javax.websocket.server.PathParam;
+import javax.websocket.server.ServerEndpoint;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
+import com.alibaba.fastjson.TypeReference;
+import com.ktg.common.core.domain.entity.SysUser;
+import com.ktg.common.core.domain.model.LoginUser;
+import com.ktg.common.utils.StringUtils;
+import com.ktg.common.utils.spring.SpringUtils;
+import com.ktg.framework.web.service.TokenService;
+import com.ktg.system.domain.SysMessage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+/**
+ * websocket 消息处理
+ * 
+ * @author ruoyi
+ */
+@Component
+@ServerEndpoint("/websocket/message/{token}")
+public class WebSocketServer
+{
+    /**
+     * WebSocketServer 日志控制器
+     */
+    private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketServer.class);
+
+
+    /**
+     * 默认最多允许同时在线人数100
+     */
+    public static int socketMaxOnlineCount = 100;
+
+    private static Semaphore socketSemaphore = new Semaphore(socketMaxOnlineCount);
+
+    /**
+     * 连接建立成功调用的方法
+     */
+    @OnOpen
+    public void onOpen(Session session,@PathParam("token") String token) throws Exception
+    {
+        boolean semaphoreFlag = false;
+        //身份验证
+        if(!StringUtils.isNotNull(token)){
+            session.close();
+            return;
+        }
+
+        TokenService tokenService = SpringUtils.getBean(TokenService.class);
+
+        LoginUser user = tokenService.getUserByToken(token);
+        if(!StringUtils.isNotNull(user)){
+            session.close();
+            return;
+        }
+
+        // 尝试获取信号量
+        semaphoreFlag = SemaphoreUtils.tryAcquire(socketSemaphore);
+        if (!semaphoreFlag)
+        {
+            // 未获取到信号量
+            LOGGER.error("\n 当前在线人数超过限制数- {}", socketMaxOnlineCount);
+            WebSocketUsers.sendMessageToUserByText(session, "当前在线人数超过限制数：" + socketMaxOnlineCount);
+            session.close();
+        }
+        else
+        {
+            // 添加用户
+            WebSocketUsers.put(user.getUsername(), session);
+            LOGGER.info("\n 建立连接 - {}", session);
+            LOGGER.info("\n 当前人数 - {}", WebSocketUsers.getUsers().size());
+            WebSocketUsers.sendMessageToUserByText(session, "连接成功");
+        }
+    }
+
+    /**
+     * 连接关闭时处理
+     */
+    @OnClose
+    public void onClose(Session session)
+    {
+        LOGGER.info("\n 关闭连接 - {}", session);
+        // 移除用户
+        WebSocketUsers.remove(session);
+        // 获取到信号量则需释放
+        SemaphoreUtils.release(socketSemaphore);
+    }
+
+    /**
+     * 抛出异常时处理
+     */
+    @OnError
+    public void onError(Session session, Throwable exception) throws Exception
+    {
+        if (session.isOpen())
+        {
+            // 关闭连接
+            session.close();
+        }
+        String sessionId = session.getId();
+        LOGGER.info("\n 连接异常 - {}", sessionId);
+        LOGGER.info("\n 异常信息 - {}", exception);
+        // 移出用户
+        WebSocketUsers.remove(session);
+        // 获取到信号量则需释放
+        SemaphoreUtils.release(socketSemaphore);
+    }
+
+    /**
+     * 服务器接收到客户端消息时调用的方法
+     */
+    @OnMessage
+    public void onMessage(String message, Session session)
+    {
+        try{
+            SysMessage msg = JSON.parseObject(message, new TypeReference<SysMessage>(){});
+            if(StringUtils.isNotNull(msg.getRecipientName())){
+                //这里必须传递username
+                WebSocketUsers.sendMesssageToUserByName(msg.getRecipientName(),message);
+            }
+        }catch (Exception e){
+            LOGGER.error("\n 错误的websocket信息格式 - {}", message);
+        }
+    }
+
+}
--- a/ktg-framework/src/main/java/com/ktg/framework/websocket/WebSocketUsers.java
+++ b/ktg-framework/src/main/java/com/ktg/framework/websocket/WebSocketUsers.java
@@ -0,0 +1,156 @@
+package com.ktg.framework.websocket;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import javax.websocket.Session;
+
+import com.ktg.common.utils.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * websocket 客户端用户集
+ * 
+ * @author ruoyi
+ */
+public class WebSocketUsers
+{
+    /**
+     * WebSocketUsers 日志控制器
+     */
+    private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketUsers.class);
+
+    /**
+     * 用户集
+     */
+    private static Map<String, Session> USERS = new ConcurrentHashMap<String, Session>();
+
+    /**
+     * 存储用户
+     *
+     * @param key 唯一键
+     * @param session 用户信息
+     */
+    public static void put(String key, Session session)
+    {
+        USERS.put(key, session);
+    }
+
+    /**
+     * 移除用户
+     *
+     * @param session 用户信息
+     *
+     * @return 移除结果
+     */
+    public static boolean remove(Session session)
+    {
+        String key = null;
+        boolean flag = USERS.containsValue(session);
+        if (flag)
+        {
+            Set<Map.Entry<String, Session>> entries = USERS.entrySet();
+            for (Map.Entry<String, Session> entry : entries)
+            {
+                Session value = entry.getValue();
+                if (value.equals(session))
+                {
+                    key = entry.getKey();
+                    break;
+                }
+            }
+        }
+        else
+        {
+            return true;
+        }
+        return remove(key);
+    }
+
+    /**
+     * 移出用户
+     *
+     * @param key 键
+     */
+    public static boolean remove(String key)
+    {
+        LOGGER.info("\n 正在移出用户 - {}", key);
+        Session remove = USERS.remove(key);
+        if (remove != null)
+        {
+            boolean containsValue = USERS.containsValue(remove);
+            LOGGER.info("\n 移出结果 - {}", containsValue ? "失败" : "成功");
+            return containsValue;
+        }
+        else
+        {
+            return true;
+        }
+    }
+
+    /**
+     * 获取在线用户列表
+     *
+     * @return 返回用户集合
+     */
+    public static Map<String, Session> getUsers()
+    {
+        return USERS;
+    }
+
+    /**
+     * 群发消息文本消息
+     *
+     * @param message 消息内容
+     */
+    public static void sendMessageToUsersByText(String message)
+    {
+        Collection<Session> values = USERS.values();
+        for (Session value : values)
+        {
+            sendMessageToUserByText(value, message);
+        }
+    }
+
+    /**
+     * 发送文本消息
+     *
+     * @param session 自己的用户名
+     * @param message 消息内容
+     */
+    public static void sendMessageToUserByText(Session session, String message)
+    {
+        if (session != null)
+        {
+            try
+            {
+                session.getBasicRemote().sendText(message);
+            }
+            catch (IOException e)
+            {
+                LOGGER.error("\n[发送消息异常]", e);
+            }
+        }
+        else
+        {
+            LOGGER.info("\n[你已离线]");
+        }
+    }
+
+    public static void sendMesssageToUserByName(String username,String message){
+        Session session = USERS.get(username);
+        if(StringUtils.isNotNull(session)){
+            try
+            {
+                session.getBasicRemote().sendText(message);
+            }
+            catch (IOException e)
+            {
+                LOGGER.error("\n[发送消息异常]", e);
+            }
+        }
+    }
+}