webSocket集群及统计在线用户数

src/main/java/com/unissoft/controller/QuartzController.java

@@ -21,6 +21,9 @@ public class QuartzController {
     @Autowired
     private QuartzService quartzService;
 
+//	@Autowired
+//    private PublishService publishService;
+
 	@ApiOperation(value = "新增任务", notes = "")
 	@PostMapping("/authority_button/add")
 	public ResultView add(QuartzModel quartz){
@@ -94,4 +97,18 @@ public class QuartzController {
 		return ResultView.error();
 	}
 
+//	@ApiOperation(value = "测试Redis发布", notes = "")
+//	@GetMapping("/authority_button/testRedisPublish")
+//	public ResultView testRedisPublish() {
+//		MessageVO vo = new MessageVO();
+//		vo.setTitle("aaa");
+//		vo.setContent("测试");
+//		vo.setFromUserId(1);
+//		vo.setToUserId(5);
+//		vo.setType(10);
+//
+//		publishService.publish(GrainConstant.MQ_MSG_CHANNEL, vo);
+//		return ResultView.success();
+//	}
+
 }

src/main/java/com/unissoft/controller/SysMessageController.java

@@ -3,15 +3,15 @@ package com.unissoft.controller;
 
 import com.unissoft.dto.MessageVO;
 import com.unissoft.enums.MessageTypeEnum;
+import com.unissoft.model.SysUserPO;
 import com.unissoft.result.ResultView;
 import com.unissoft.service.SysMessageService;
+import com.unissoft.utils.UserUtils;
 import io.swagger.annotations.ApiOperation;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.*;
 
-import java.util.Collection;
-
 /**
  * <p>
  * 消息推送表 前端控制器
@@ -26,15 +26,13 @@ import java.util.Collection;
 public class SysMessageController {
 
     @Autowired
-    private WebSocketServer webSocketServer;
-
-    @Autowired
     private SysMessageService sysMessageService;
 
-
     @ApiOperation(value = "向用户发送信息", notes = "")
     @PostMapping("/authority/sendMessage")
     public ResultView sendMessage(@RequestBody MessageVO messageVO) {
+        SysUserPO userPO = UserUtils.getCurrUserInfo();
+        messageVO.setFromUserId(userPO.getuId());
         if(null == messageVO.getType()) {
             messageVO.setType(MessageTypeEnum.ATTENTION.getCode());//默认类型
         }
@@ -42,23 +40,10 @@ public class SysMessageController {
         return ResultView.success();
     }
 
-    /**
-     * @TODO 后期需改造成 获取所有用户ID，并消息持久化
-     * @param messageVO
-     * @return
-     */
-    @ApiOperation(value = "向所有用户发送信息", notes = "")
-    @PostMapping("/authority/sendAllUsersMessage")
-    public ResultView sendAllUsersMessage(@RequestBody MessageVO messageVO) {
-        Collection<Integer> userIds = webSocketServer.getOnlineUserIds();
-        messageVO.setToUserIds(userIds);
-        return sendMessage(messageVO);
-    }
-
     @ApiOperation(value = "根据用户id删除webSocket连接", notes = "")
     @GetMapping("/authority/removeSession/{id}")
     public ResultView removeSession(@PathVariable int id) {
-        webSocketServer.removeSession(id);
+        sysMessageService.removeWebSocketSession(id);
         return ResultView.success();
     }
 

src/main/java/com/unissoft/controller/WebSocketServer.java

@@ -1,29 +1,29 @@
 package com.unissoft.controller;
 
-import com.alibaba.fastjson.JSON;
 import com.unissoft.config.MyApplicationContextAware;
+import com.unissoft.constant.GrainConstant;
 import com.unissoft.dto.MessageVO;
+import com.unissoft.enums.OnlineStatusEnum;
+import com.unissoft.redismq.PublishService;
 import com.unissoft.service.SysMessageService;
+import com.unissoft.utils.JsonUtil;
+import com.unissoft.vo.UserOnlineStatusVO;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Component;
 
 import javax.websocket.*;
 import javax.websocket.server.PathParam;
 import javax.websocket.server.ServerEndpoint;
 import java.io.IOException;
-import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicInteger;
 
 @Slf4j
 @ServerEndpoint("/webSocket/{userId}")
 @Component
 public class WebSocketServer {
 
-    /** 静态变量，用来记录当前在线连接数(把它设计成线程安全) */
-    private static AtomicInteger onlineNum = new AtomicInteger(0);
-
     /** 用来存放每个客户端对应的WebSocket的Session对象 */
     private static ConcurrentHashMap<Integer, Session> sessionPools = new ConcurrentHashMap<>();
 
@@ -34,41 +34,43 @@ public class WebSocketServer {
      */
     @OnOpen
     public void onOpen(Session session, @PathParam(value = "userId") Integer userId) {
-
-        log.error("------------------用户ID：" + userId);
-
         if(sessionPools.containsKey(userId)) {
             sessionPools.remove(userId);
-            sessionPools.put(userId, session);
-        } else {
-            sessionPools.put(userId, session);
-            addOnlineCount();
-            log.info("有新连接加入：{}，当前在线人数为：{}", "userId为" + userId, onlineNum.get());
         }
+        sessionPools.put(userId, session);
+        addOnlineUser(userId);
+        publishStatisticsOnlineStatus();//发布在线统计状态
 
         //登录后，查询是否有待推送的通知，如果有立马推送
         new Thread() {
             @Override
             public void run() {
-                pushMessage2User(userId);
+                pushMessage2User(userId, session);
             }
         }.start();
     }
 
-    private void pushMessage2User(Integer userId) {
+    private void pushMessage2User(Integer userId, Session session) {
         SysMessageService sysMessageService = MyApplicationContextAware.getBean(SysMessageService.class);
 
         List<MessageVO> list = sysMessageService.getUnpushedUserMessage(userId);
-        if(null != list && ! list.isEmpty()) {
+        if(null != list && list.isEmpty() == false) {
             for(MessageVO vo : list) {
-                String content = com.alibaba.fastjson.JSONObject.toJSONString(vo);
-                sendUserMessage(userId, content);
+                String content = JsonUtil.parseObjToJson(vo);
+                sendMessage(session, content);
             }
             //设置为已推送状态
             sysMessageService.batchSetPushedStatus(list);
         }
     }
 
+    private void publishStatisticsOnlineStatus() {
+        PublishService publishService = MyApplicationContextAware.getBean(PublishService.class);
+        UserOnlineStatusVO onlineStatusVO = new UserOnlineStatusVO();
+        onlineStatusVO.setOnlineStatus(OnlineStatusEnum.STATISTICS.getCode());
+        publishService.publish(GrainConstant.MQ_USER_STATUS_CHANNEL, onlineStatusVO);
+    }
+
 
     /**
      * 关闭连接时调用
@@ -77,7 +79,14 @@ public class WebSocketServer {
     @OnClose
     public void onClose(@PathParam(value = "userId") Integer userId) {
         removeSession(userId);
-        log.info("有一连接关闭：{}，当前在线人数为：{}", "userId为" + userId, onlineNum.get());
+        publishStatisticsOnlineStatus();//发布在线统计状态
+    }
+
+    public static void removeSession(Integer userId) {
+        if(sessionPools.containsKey(userId)) {
+            sessionPools.remove(userId);
+        }
+        subOnlineUser(userId);
     }
 
     /**
@@ -86,16 +95,11 @@ public class WebSocketServer {
      */
     @OnMessage
     public void onMessage(String message, @PathParam(value = "userId") Integer userId) {
-        log.info("服务端收到客户端[{}]的消息:{}", userId, message);
-
-        MessageVO messageVO = JSON.parseObject(message, MessageVO.class);
+        MessageVO messageVO = JsonUtil.parseJsonToObj(message, MessageVO.class);
         if(null != messageVO) {
-            if(messageVO.isBroadcast()) {//广播
-                sendAllUsersMessage(messageVO.getContent());
-            } else {
-                Collection<Integer> toUserIds = messageVO.getToUserIds();
-                sendBatchUsersMessage(toUserIds, messageVO.getContent());
-            }
+            messageVO.setFromUserId(userId);
+            SysMessageService sysMessageService = MyApplicationContextAware.getBean(SysMessageService.class);
+            sysMessageService.sendMessage(messageVO);
         }
     }
 
@@ -109,17 +113,6 @@ public class WebSocketServer {
         log.error("发生错误，错误原因：" + error.getMessage());
     }
 
-    /**
-     * 删除连接
-     * @param userId
-     */
-    public void removeSession(Integer userId) {
-        if(sessionPools.containsKey(userId)) {
-            sessionPools.remove(userId);
-            subOnlineCount();
-        }
-    }
-
 
     /**
      * 向指定用户发送信息
@@ -131,72 +124,60 @@ public class WebSocketServer {
         if(null == session) {//用户不在线，返回false
             return false;
         }
-        try {
-            sendMessage(session, content);
-            return true;
-        } catch (Exception e) {//推送失败，返回false
-            log.error(e.getMessage());
-            return false;
-        }
+        return sendMessage(session, content);
     }
 
-    /**
-     * 向所有在线用户发送信息
-     * @param content
-     */
-    public void sendAllUsersMessage(String content) {
-        if(! sessionPools.isEmpty()) {
-            for(Session session : sessionPools.values()) {
-                try {
-                    sendMessage(session, content);
-                } catch (Exception e) {
-                    log.error(e.getMessage());
-                }
-            }
+    public void noticeOnlineNum2All(long onlineNum) {
+        MessageVO messageVO = new MessageVO();
+        messageVO.setType(9999);//表示在线用户数信息
+        messageVO.setOnlineNum(onlineNum);
+        String content = JsonUtil.parseObjToJson(messageVO);
+        for(Session toSession : sessionPools.values()) {
+            sendMessage(toSession, content);
         }
     }
 
-    /**
-     * 向某批用户在线发送信息
-     * @param toUserIds
-     * @param content
-     */
-    public void sendBatchUsersMessage(Collection<Integer> toUserIds, String content) {
-        if(null != toUserIds && ! toUserIds.isEmpty()) {
-            for(Integer toUserId : toUserIds) {
-                sendUserMessage(toUserId, content);
-            }
-        }
-    }
 
     /**
      * 向某客户端发送消息
      * @param toSession   接收客户端session
-     * @param content   信息内空
+     * @param content   信息内容
      * @throws IOException
      */
-    private synchronized void sendMessage(Session toSession, String content) throws IOException {
-        if (toSession != null) {
+    private synchronized boolean sendMessage(Session toSession, String content) {
+        try {
             toSession.getBasicRemote().sendText(content);
+            return true;
+        } catch (IOException e) {//推送失败，返回false
+            log.error(e.getMessage());
+            return false;
         }
     }
 
 
     /**
-     * 在线用户量加1
+     * 在线用户添加
      */
-    public static void addOnlineCount() {
-        onlineNum.incrementAndGet();
+    public static void addOnlineUser(Integer userId) {
+        RedisTemplate redisTemplate = MyApplicationContextAware.getBean(RedisTemplate.class);
+        redisTemplate.opsForSet().add(GrainConstant.ONLINE_USERS, userId);
     }
 
     /**
-     * 在线用户量减1
+     * 在线用户删除
      */
-    public static void subOnlineCount() {
-        onlineNum.decrementAndGet();
+    public static void subOnlineUser(Integer userId) {
+        RedisTemplate redisTemplate = MyApplicationContextAware.getBean(RedisTemplate.class);
+        redisTemplate.opsForSet().remove(GrainConstant.ONLINE_USERS, userId);
     }
 
-    public static Collection<Integer> getOnlineUserIds() {
-        return sessionPools.keySet();
+    /**
+     * 是否在该服务上存在连接的用户
+     * @param userId
+     * @return
+     */
+    public static boolean containsUser(Integer userId) {
+        return sessionPools.containsKey(userId);
     }
+
 }

src/main/java/com/unissoft/enums/OnlineStatusEnum.java

@@ -0,0 +1,18 @@
+package com.unissoft.enums;
+
+import lombok.Getter;
+
+@Getter
+public enum OnlineStatusEnum {
+    STATISTICS(1, "统计"),
+    OFFLINE(0, "下线"),
+    ;
+
+    private Integer code;
+    private String desc;
+
+    OnlineStatusEnum(Integer code, String desc) {
+        this.code = code;
+        this.desc = desc;
+    }
+}

src/main/java/com/unissoft/job/ToBeDoneJob.java

@@ -17,7 +17,10 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 import java.io.Serializable;
-import java.util.*;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 
 @DisallowConcurrentExecution
 @Component
@@ -72,9 +75,8 @@ public class ToBeDoneJob implements Job, Serializable {
                 String content = vo.getDepotName() + "-" + vo.getHouseName() + "-" + "待处理";
                 messageVO.setContent(content);
                 //
-                Set<Integer> userIdSet = new HashSet<>();
-                userIdSet.add(vo.getUserId());
-                messageVO.setToUserIds(userIdSet);
+                messageVO.setFromUserId(-1);//-1 代表系统推送
+                messageVO.setToUserId(vo.getUserId());
                 messageVO.setType(MessageTypeEnum.ATTENTION.getCode());
                 sysMessageService.sendMessage(messageVO);
             }

src/main/java/com/unissoft/mapper/SysMessageMapper.java

@@ -1,9 +1,13 @@
 package com.unissoft.mapper;
 
 import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.unissoft.dto.MessageVO;
 import com.unissoft.model.SysMessage;
+import org.apache.ibatis.annotations.Param;
 import org.springframework.stereotype.Repository;
 
+import java.util.List;
+
 /**
  * <p>
  * 消息推送表 Mapper 接口
@@ -15,4 +19,7 @@ import org.springframework.stereotype.Repository;
 @Repository
 public interface SysMessageMapper extends BaseMapper<SysMessage> {
 
+    void batchSetPushedStatus(List<MessageVO> list);
+
+    void setPushedStatus(@Param("id") Integer id);
 }

src/main/java/com/unissoft/mapper/SysMessageMapper.xml

@@ -8,7 +8,8 @@
         <result column="type" property="type" />
         <result column="title" property="title" />
         <result column="content" property="content" />
-        <result column="user_id" property="userId" />
+        <result column="from_user_id" property="fromUserId" />
+        <result column="to_user_id" property="toUserId" />
         <result column="push_status" property="pushStatus" />
         <result column="url" property="url" />
         <result column="create_by" property="createBy" />
@@ -21,7 +22,24 @@
 
     <!-- 通用查询结果列 -->
     <sql id="Base_Column_List">
-        id, type, title, content, user_id, push_status, url, create_by, create_time, update_by, update_time, delete_by, delete_time
+        id, type, title, content, from_user_id, to_user_id, push_status, url,
+        create_by, create_time, update_by, update_time, delete_by, delete_time
     </sql>
 
+    <update id="batchSetPushedStatus" parameterType="list">
+        update sys_message
+        set push_status = 1
+        <where>
+            <foreach item="item" collection="list"  index="index" separator=" or " open="(" close=")">
+                id = #{item.id}
+            </foreach>
+        </where>
+    </update>
+
+    <update id="setPushedStatus" parameterType="Integer">
+        update sys_message
+        set push_status = 1
+        where id = #{id}
+    </update>
+
 </mapper>

src/main/java/com/unissoft/mapper/SysResourceRoleMapper.java

@@ -4,6 +4,7 @@ import com.baomidou.mybatisplus.core.mapper.BaseMapper;
 import com.unissoft.dto.ResourceRoleInfoDto;
 import com.unissoft.model.SysResourceRole;
 import org.apache.ibatis.annotations.Param;
+import org.springframework.stereotype.Repository;
 
 import java.util.List;
 
@@ -15,6 +16,7 @@ import java.util.List;
  * @author wsm
  * @since 2020-11-25
  */
+@Repository
 public interface SysResourceRoleMapper extends BaseMapper<SysResourceRole> {
 
     /**

src/main/java/com/unissoft/model/SysMessage.java

@@ -37,9 +37,13 @@ public class SysMessage extends BaseModel implements Serializable {
     @TableField("content")
     private String content;
 
-    @ApiModelProperty(value = "用户id")
-    @TableField("user_id")
-    private Integer userId;
+    @ApiModelProperty(value = "发送用户ID")
+    @TableField("from_user_id")
+    private Integer fromUserId;
+
+    @ApiModelProperty(value = "接收用户ID")
+    @TableField("to_user_id")
+    private Integer toUserId;
 
     @ApiModelProperty(value = "推送状态，0:未推送,  1:已推送")
     @TableField("push_status")

src/main/java/com/unissoft/redismq/MsgListenerHandle.java

@@ -0,0 +1,58 @@
+package com.unissoft.redismq;
+
+import com.unissoft.controller.WebSocketServer;
+import com.unissoft.dto.MessageVO;
+import com.unissoft.service.SysMessageService;
+import com.unissoft.utils.JsonUtil;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.connection.Message;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
+import org.springframework.stereotype.Component;
+
+/**
+ * mqMsgChannel 订阅频道处理类
+ */
+@Slf4j
+@Component
+public class MsgListenerHandle extends MessageListenerAdapter {
+
+    @Autowired
+    private RedisTemplate<String, String> redisTemplate;
+
+    @Autowired
+    private WebSocketServer webSocketServer;
+
+    @Autowired
+    private SysMessageService sysMessageService;
+
+    /**
+     * 收到监听消息
+     * @param message
+     * @param bytes
+     */
+    @Override
+    public void onMessage(Message message, byte[] bytes) {
+        byte[] body = message.getBody();
+        byte[] channel = message.getChannel();
+        String messageStr;
+        String topic;
+        try {
+            messageStr = redisTemplate.getStringSerializer().deserialize(body);
+            topic = redisTemplate.getStringSerializer().deserialize(channel);
+            log.info("收到主题:" + topic + ", 信息内容：" + messageStr);
+        } catch (Exception e) {
+            log.error(e.getMessage(), e);
+            return;
+        }
+
+        MessageVO messageVO = JsonUtil.parseJsonToObj(messageStr, MessageVO.class);
+        if(WebSocketServer.containsUser(messageVO.getToUserId())) {
+            boolean pushedStatus = webSocketServer.sendUserMessage(messageVO.getToUserId(), messageStr);
+            if(pushedStatus) {
+                sysMessageService.setPushedStatus(messageVO.getId());
+            }
+        }
+    }
+}

src/main/java/com/unissoft/redismq/PublishService.java

@@ -0,0 +1,22 @@
+package com.unissoft.redismq;
+
+import com.unissoft.utils.JsonUtil;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.stereotype.Component;
+
+@Component
+public class PublishService {
+	
+	@Autowired
+	StringRedisTemplate redisTemplate;
+
+	/**
+	 * 发布方法
+	 * @param channel 消息发布订阅 主题
+	 * @param message 消息信息
+	 */
+	public void publish(String channel, Object message) {
+		redisTemplate.convertAndSend(channel, JsonUtil.parseObjToJson(message));
+	}
+}

src/main/java/com/unissoft/redismq/RedisListenerBean.java

@@ -0,0 +1,34 @@
+package com.unissoft.redismq;
+
+import com.unissoft.constant.GrainConstant;
+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.PatternTopic;
+import org.springframework.data.redis.listener.RedisMessageListenerContainer;
+
+@Configuration
+public class RedisListenerBean {
+
+    /**
+     * redis消息监听器容器
+     * 可以添加多个监听不同话题的redis监听器，只需要把消息监听器和相应的消息订阅处理器绑定，该消息监听器
+     * 通过反射技术调用消息订阅处理器的相关方法进行一些业务处理
+     * @param connectionFactory
+     * @param msgListenerHandle
+     * @param userStatusListenerHandle
+     * @return
+     */
+    @Bean
+    public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
+                                                   MsgListenerHandle msgListenerHandle,
+                                                   UserStatusListenerHandle userStatusListenerHandle){
+        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
+        container.setConnectionFactory(connectionFactory);
+        //订阅频道，可订阅多个
+        container.addMessageListener(msgListenerHandle, new PatternTopic(GrainConstant.MQ_MSG_CHANNEL));
+        container.addMessageListener(userStatusListenerHandle, new PatternTopic(GrainConstant.MQ_USER_STATUS_CHANNEL));
+        return container;
+    }
+
+}

src/main/java/com/unissoft/redismq/UserStatusListenerHandle.java

@@ -0,0 +1,60 @@
+package com.unissoft.redismq;
+
+import com.unissoft.constant.GrainConstant;
+import com.unissoft.controller.WebSocketServer;
+import com.unissoft.enums.OnlineStatusEnum;
+import com.unissoft.utils.JsonUtil;
+import com.unissoft.vo.UserOnlineStatusVO;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.connection.Message;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
+import org.springframework.stereotype.Component;
+
+/**
+ * mqUserStatusChannel 订阅频道处理类
+ */
+@Slf4j
+@Component
+public class UserStatusListenerHandle extends MessageListenerAdapter {
+
+    @Autowired
+    private RedisTemplate<String, String> redisTemplate;
+
+    @Autowired
+    private WebSocketServer webSocketServer;
+
+    /**
+     * 收到监听消息
+     * @param message
+     * @param bytes
+     */
+    @Override
+    public void onMessage(Message message, byte[] bytes) {
+        byte[] body = message.getBody();
+        byte[] channel = message.getChannel();
+        String rawMsg;
+        String topic;
+        try {
+            rawMsg = redisTemplate.getStringSerializer().deserialize(body);
+            topic = redisTemplate.getStringSerializer().deserialize(channel);
+            log.info("收到主题:" + topic + ", 信息内容：" + rawMsg);
+        } catch (Exception e) {
+            log.error(e.getMessage(), e);
+            return;
+        }
+        //
+        UserOnlineStatusVO onlineStatusVO = JsonUtil.parseJsonToObj(rawMsg, UserOnlineStatusVO.class);
+        if(onlineStatusVO.getOnlineStatus() == OnlineStatusEnum.OFFLINE.getCode()) {
+            if(WebSocketServer.containsUser(onlineStatusVO.getUserId())) {
+                WebSocketServer.removeSession(onlineStatusVO.getUserId());
+            }
+        } else if(onlineStatusVO.getOnlineStatus() == OnlineStatusEnum.STATISTICS.getCode()) {//统计在线人数
+            long onlineNum = redisTemplate.opsForSet().size(GrainConstant.ONLINE_USERS);//在线人数
+            //通知所有用户 在线人数
+            webSocketServer.noticeOnlineNum2All(onlineNum);
+        }
+
+    }
+}

src/main/java/com/unissoft/service/SysMessageService.java

@@ -5,6 +5,7 @@ import com.unissoft.dto.MessageVO;
 import com.unissoft.model.SysMessage;
 
 import java.util.List;
+import java.util.Set;
 
 /**
  * <p>
@@ -18,8 +19,13 @@ public interface SysMessageService extends IService<SysMessage> {
 
     void sendMessage(MessageVO messageVO);
 
+    void batchSendMessage(MessageVO messageVO, Set<Integer> userIds);
+
+    void setPushedStatus(Integer id);
+
     List<MessageVO> getUnpushedUserMessage(Integer userId);
 
     void batchSetPushedStatus(List<MessageVO> voList);
 
+    void removeWebSocketSession(int id);
 }

src/main/java/com/unissoft/service/impl/SysAnnouncementServiceImpl.java

@@ -194,15 +194,15 @@ public class SysAnnouncementServiceImpl extends ServiceImpl<SysAnnouncementMappe
      */
     private void saveAnnouncementUsersAndSendMessage(SysAnnouncement model, SysUserPO userPO) {
         Map<Integer, String> users = getNoticeUserInfo(userPO.getDepotId(), model.getRecvType(), model.getRecvDetails());
-        if(null != users && ! users.isEmpty()) {
+        if(null != users && users.isEmpty() == false) {
             saveAnnouncementUsers(model.getId(), users);
 
             MessageVO messageVO = new MessageVO();
             messageVO.setTitle(model.getTitle());
             messageVO.setContent(model.getTitle());
             messageVO.setType(MessageTypeEnum.ATTENTION.getCode());
-            messageVO.setToUserIds(users.keySet());
-            sysMessageService.sendMessage(messageVO);
+            messageVO.setFromUserId(userPO.getuId());
+            sysMessageService.batchSendMessage(messageVO, users.keySet());
         }
     }
 

src/main/java/com/unissoft/service/impl/SysMessageServiceImpl.java

@@ -1,21 +1,24 @@
 package com.unissoft.service.impl;
 
-import com.alibaba.fastjson.JSONObject;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
-import com.unissoft.controller.WebSocketServer;
+import com.unissoft.constant.GrainConstant;
 import com.unissoft.dto.MessageVO;
+import com.unissoft.enums.OnlineStatusEnum;
 import com.unissoft.mapper.SysMessageMapper;
 import com.unissoft.model.SysMessage;
+import com.unissoft.redismq.PublishService;
 import com.unissoft.service.SysMessageService;
+import com.unissoft.vo.UserOnlineStatusVO;
+import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
 import java.util.ArrayList;
-import java.util.Collection;
 import java.util.Date;
 import java.util.List;
+import java.util.Set;
 
 /**
  * <p>
@@ -32,33 +35,36 @@ public class SysMessageServiceImpl extends ServiceImpl<SysMessageMapper, SysMess
     private SysMessageMapper sysMessageMapper;
 
     @Autowired
-    private WebSocketServer webSocketServer;
+    private PublishService publishService;
 
     @Override
     @Transactional
     public void sendMessage(MessageVO messageVO) {
-        Collection<Integer> toUserIds = messageVO.getToUserIds();
-        if(null != toUserIds && toUserIds.isEmpty() == false) {
-            messageVO.setToUserIds(null);
+        if(null != messageVO.getToUserId()) {
             SysMessage message = new SysMessage();
-            message.setType(messageVO.getType());
-            message.setTitle(messageVO.getTitle());
-            message.setContent(messageVO.getContent());
-
+            BeanUtils.copyProperties(messageVO, message);
+            message.setPushStatus(false);
             Date now = new Date();
             message.setCreateTime(now);
             message.setUpdateTime(now);
-            for(Integer toUserId : toUserIds) {
-                messageVO.setUserId(toUserId);
-                String content = JSONObject.toJSONString(messageVO);
-                boolean pushStatus = webSocketServer.sendUserMessage(toUserId, content);
-                message.setUserId(toUserId);
-                message.setPushStatus(pushStatus);
-                sysMessageMapper.insert(message);
-            }
+            //
+            sysMessageMapper.insert(message);
+            messageVO.setId(message.getId());
+            //
+            publishService.publish(GrainConstant.MQ_MSG_CHANNEL, messageVO);
         }
     }
 
+    @Override
+    @Transactional
+    public void batchSendMessage(MessageVO messageVO, Set<Integer> userIds) {
+        if(null != userIds && userIds.isEmpty() == false) {
+            for(Integer userId : userIds) {
+                messageVO.setToUserId(userId);
+                sendMessage(messageVO);
+            }
+        }
+    }
 
     /**
      * 获取未推送的用户消息
@@ -69,7 +75,7 @@ public class SysMessageServiceImpl extends ServiceImpl<SysMessageMapper, SysMess
     public List<MessageVO> getUnpushedUserMessage(Integer userId) {
         List<MessageVO> result = null;
         QueryWrapper<SysMessage> ew = new QueryWrapper<>();
-        ew.eq("user_id", userId);
+        ew.eq("to_user_id", userId);
         ew.eq("push_status", 0);
         List<SysMessage> list = sysMessageMapper.selectList(ew);
 
@@ -77,11 +83,7 @@ public class SysMessageServiceImpl extends ServiceImpl<SysMessageMapper, SysMess
             result = new ArrayList<>();
             for(SysMessage sysMessage : list) {
                 MessageVO messageVO = new MessageVO();
-                messageVO.setId(sysMessage.getId());
-                messageVO.setUserId(userId);
-                messageVO.setTitle(sysMessage.getTitle());
-                messageVO.setContent(sysMessage.getContent());
-                messageVO.setType(sysMessage.getType());
+                BeanUtils.copyProperties(sysMessage, messageVO);
                 result.add(messageVO);
             }
         }
@@ -89,13 +91,22 @@ public class SysMessageServiceImpl extends ServiceImpl<SysMessageMapper, SysMess
     }
 
     @Override
-    public void batchSetPushedStatus(List<MessageVO> voList) {
-        List<SysMessage> list = new ArrayList<>();
-        for(MessageVO vo : voList) {
-            SysMessage sysMessage = getById(vo.getId());
-            sysMessage.setPushStatus(true);
-            list.add(sysMessage);
-        }
-        updateBatchById(list);
+    public void batchSetPushedStatus(List<MessageVO> list) {
+        sysMessageMapper.batchSetPushedStatus(list);
     }
+
+    @Override
+    public void setPushedStatus(Integer id) {
+        sysMessageMapper.setPushedStatus(id);
+    }
+
+    @Override
+    public void removeWebSocketSession(int id) {
+        UserOnlineStatusVO onlineStatusVO = new UserOnlineStatusVO();
+        onlineStatusVO.setUserId(id);
+        onlineStatusVO.setOnlineStatus(OnlineStatusEnum.OFFLINE.getCode());
+        publishService.publish(GrainConstant.MQ_USER_STATUS_CHANNEL, onlineStatusVO);
+    }
+
+
 }

src/main/java/com/unissoft/service/impl/SysNoticeUserServiceImpl.java

@@ -5,7 +5,6 @@ import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
 import com.google.common.collect.Maps;
 import com.unissoft.MyConstant;
 import com.unissoft.constant.GrainConstant;
-import com.unissoft.controller.WebSocketServer;
 import com.unissoft.dto.MessageVO;
 import com.unissoft.dto.SendNoticeVO;
 import com.unissoft.dto.SysNoticeVO;
@@ -41,8 +40,6 @@ import java.util.Map;
 @Service
 public class SysNoticeUserServiceImpl extends ServiceImpl<SysNoticeUserMapper, SysNoticeUser> implements SysNoticeUserService {
 
-
-
     @Autowired
     private SysNoticeUserMapper noticeUserMapper;
 
@@ -52,9 +49,6 @@ public class SysNoticeUserServiceImpl extends ServiceImpl<SysNoticeUserMapper, S
     @Autowired
     private SysMessageService sysMessageService;
 
-    @Autowired
-    private WebSocketServer webSocketServer;
-
     @Override
     public Page<SysNoticeUserVO> getPage(PageParam pageParam, Integer userId) {
         Integer pageIndex = pageParam.getPageIndex();
@@ -142,17 +136,16 @@ public class SysNoticeUserServiceImpl extends ServiceImpl<SysNoticeUserMapper, S
                     save(sysNoticeUser);
                 }
                 //
-                MessageVO messageVO = getMessageVO(users, sysNotice);
-                sysMessageService.sendMessage(messageVO);
+                MessageVO messageVO = getMessageVO(sysNotice);
+                sysMessageService.batchSendMessage(messageVO, users.keySet());
             }
         }
 
     }
 
-    private MessageVO getMessageVO(Map<Integer, String> users, SysNotice sysNotice) {
+    private MessageVO getMessageVO(SysNotice sysNotice) {
 
         MessageVO messageVO = new MessageVO();
-        messageVO.setToUserIds(users.keySet());
         messageVO.setTitle(sysNotice.getTitle());
         messageVO.setContent(sysNotice.getContent());
 

src/main/java/com/unissoft/vo/UserOnlineStatusVO.java

@@ -0,0 +1,13 @@
+package com.unissoft.vo;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+import java.io.Serializable;
+
+@Data
+@EqualsAndHashCode(callSuper = false)
+public class UserOnlineStatusVO implements Serializable {
+    private Integer userId;
+    private Integer onlineStatus;
+}