devbox
二进制舞者2
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

(二十)ATP应用测试平台——websocket实现微服务版在线客服聊天室实战案例_微服务聊天系统

作者：二进制舞者2 | 2024-01-30 21:39:20

赞

踩

微服务聊天系统

前言

在前面的博客内容中我们介绍了如何使用websocket实现一个网页版的在线客服聊天室，众所周知，由于websocket是一个长连接，要和服务端保持会话连接，所以其本身并不适用于微服务环境，在微服务环境中，有可能A、B俩个客户端连接到不同的服务A、B中，这样就没法保证A、B俩个客户端完成聊天的功能，因为会话不在同一台服务器上，A、B无法感知到对方发送的消息，为了解决websocket单机的这个痛点，我们引入消息中间键RocketMQ的广播机制，实现消息的转发，从而实现微服务版的websocke聊天室功能。其架构如下：

本节内容使用的主要技术包含springboot、redis、rocketmq、vue等，关于中间键的搭建本节内容不在展开，请关注作者的往期博客内容。 

正文

  • 引入websocket、redis和rocketmq的pom依赖

①核心pom依赖

  1.  <!--           rocketmq-->
  2. <dependency>
  3. 	<groupId>org.apache.rocketmq</groupId>
  4. 	<artifactId>rocketmq-spring-boot-starter</artifactId>
  5. 	<version>2.2.2</version>
  6. </dependency>
  7. <!--           websocket-->
  8. <dependency>
  9. 	<groupId>org.springframework.boot</groupId>
  10. 	<artifactId>spring-boot-starter-websocket</artifactId>
  11. </dependency>
  12. <!--           redis-->
  13. <dependency>
  14. 	<groupId>org.springframework.session</groupId>
  15. 	<artifactId>spring-session-data-redis</artifactId>
  16. 	<version>2.4.3</version>
  17. </dependency>

 PS：可以按需引入自己需要的依赖，作者这里只列出核心的pom依赖

  • 配置application.yml

①配置文件

  1. server:
  2.   port: 8888
  3. spring:
  4.   #数据源配置
  5.   datasource:
  6.     dynamic:
  7.       primary: master #设置默认的数据源或者数据源组,默认值即为master
  8.       strict: false #设置严格模式,默认false不启动. 启动后在未匹配到指定数据源时候会抛出异常,不启动则使用默认数据源.
  9.       datasource:
  10.         master:
  11.           url: jdbc:mysql://192.168.56.10:3306/atp
  12.           username: root
  13.           password: root
  14.           driver-class-name: com.mysql.cj.jdbc.Driver # 3.2.0开始支持SPI可省略此配置
  15.   profiles:
  16.     active: dev
  17.   servlet:
  18.     multipart:
  19.       max-file-size: 52428800
  20.       max-request-size: 52428800
  21.  
  22.   #redis配置
  23.   redis:
  24.     #redisson配置
  25.     redisson:
  26.       file: classpath:redisson.yaml
  27.     #默认数据分区
  28.     database: 0
  29.     #redis集群节点配置
  30.     cluster:
  31.       nodes:
  32.         - 192.168.56.10:6379
  33.         - 192.168.56.10:6380
  34.         - 192.168.56.10.6381
  35.       max-redirects: 3
  36.     #超时时间
  37.     timeout: 10000
  38.     #哨兵节点配置
  39.     sentinel:
  40.       master: mymaster
  41.       nodes:
  42.         - "192.168.56.10:26379"
  43.         - "192.168.56.10:26380"
  44.         - "192.168.56.10:26381"
  45.     #redis密码
  46.     password: root
  47.     #redis 客户端工具
  48.     lettuce:
  49.       pool:
  50.         # 连接池最大连接数(使用负值表示没有限制) 默认为8
  51.         max-active: 8
  52.         # 连接池中的最小空闲连接 默认为 0
  53.         min-idle: 1
  54.         # 连接池最大阻塞等待时间(使用负值表示没有限制) 默认为-1
  55.         max-wait: 1000
  56.         # 连接池中的最大空闲连接 默认为8
  57.         max-idle: 8
  58.   session:
  59.     store-type: redis
  60.     redis:
  61.       flush-mode: on_save
  62.       namespace: spring:session:atp
  63.   thymeleaf:
  64.     cache: false
  65.  
  66. #mybatisplus配置
  67. mybatis-plus:
  68.   mapper-locations: classpath*:/mapper/*/*Mapper.xml
  69.   type-aliases-package: com.yundi.atp.platform.module.*.entity
  70.   configuration:
  71.     map-underscore-to-camel-case: true
  72.   global-config:
  73.     db-config:
  74.       id-type: assign_id
  75.  
  76. #rocketmq配置
  77. rocketmq:
  78.   #注册地址
  79.   name-server: 192.168.56.10:9876;192.168.56.10:9877
  80.   producer:
  81.     #生产者组名称
  82.     group: atp-producer
  83.     #命名空间
  84.     namespace: atp
  85.     #异步消息发送失败重试次数，默认是2
  86.     retry-times-when-send-async-failed: 2
  87.     #发送消息超时时间，默认2000ms
  88.     send-message-timeout: 2000
  89.     #消息的最大长度：默认1024 * 1024 * 4(默认4M)
  90.     max-message-size: 40000000
  91.     #压缩消息阈值，超过4k就压缩
  92.     compress-message-body-threshold: 4096
  93.     #是否发送失败，重试另外的broker
  94.     retry-next-server: false
  95.     #是否启用消息追踪
  96.     enable-msg-trace: false
  97.     #默认追踪的主题
  98.     customized-trace-topic: RMQ_SYS_TRACE_TOPIC
  99.     #消息发送失败重试的次数
  100.     retry-times-when-send-failed: 2

  • 创建websocket服务配置WebSocketConfig.java

  1. package com.yundi.atp.platform.websocket;
  2.  
  3. import org.springframework.context.annotation.Bean;
  4. import org.springframework.context.annotation.Configuration;
  5. import org.springframework.web.socket.server.standard.ServerEndpointExporter;
  6.  
  7. @Configuration
  8. public class WebSocketConfig {
  9.     /**
  10.      * 注入ServerEndpointExporter，
  11.      * 这个bean会自动注册使用了@ServerEndpoint注解声明的Websocket endpoint
  12.      */
  13.     @Bean
  14.     public ServerEndpointExporter serverEndpointExporter() {
  15.         return new ServerEndpointExporter();
  16.     }
  17. }
 

 

  •  
    创建微服务版websocket服务
     
 

 
 
  1. package com.yundi.atp.platform.websocket;
  2.  
  3. import com.alibaba.fastjson.JSON;
  4. import com.yundi.atp.platform.common.Constant;
  5. import com.yundi.atp.platform.enums.MessageType;
  6. import com.yundi.atp.platform.module.test.entity.ChatMsg;
  7. import com.yundi.atp.platform.module.test.service.ChatMsgService;
  8. import com.yundi.atp.platform.rocketmq.RocketConstant;
  9. import com.yundi.atp.platform.rocketmq.RocketProducer;
  10. import lombok.extern.slf4j.Slf4j;
  11. import org.springframework.beans.factory.annotation.Autowired;
  12. import org.springframework.data.redis.core.RedisTemplate;
  13. import org.springframework.stereotype.Component;
  14.  
  15. import javax.websocket.OnClose;
  16. import javax.websocket.OnMessage;
  17. import javax.websocket.OnOpen;
  18. import javax.websocket.Session;
  19. import javax.websocket.server.PathParam;
  20. import javax.websocket.server.ServerEndpoint;
  21. import java.util.Map;
  22. import java.util.Set;
  23. import java.util.UUID;
  24. import java.util.concurrent.ConcurrentHashMap;
  25. import java.util.concurrent.CopyOnWriteArraySet;
  26.  
  27.  
  28. @Slf4j
  29. @Component
  30. @ServerEndpoint(Constant.WEBSOCKET_MQ_URL + "{userName}")
  31. public class WebSocketMqServer {
  32.     /**
  33.      * 会话session
  34.      */
  35.     private Session session;
  36.  
  37.     /**
  38.      * socket连接
  39.      */
  40.     private static CopyOnWriteArraySet<WebSocketMqServer> webSockets = new CopyOnWriteArraySet<>();
  41.  
  42.     /**
  43.      * 会话连接池
  44.      */
  45.     private static Map<String, Session> sessionPool = new ConcurrentHashMap<>();
  46.  
  47.     /**
  48.      * 消息持久化
  49.      */
  50.     private static ChatMsgService chatMsgService;
  51.     /**
  52.      * redis
  53.      */
  54.     private static RedisTemplate redisTemplate;
  55.  
  56.     /**
  57.      * RocketMQ消息工具类
  58.      */
  59.     private static RocketProducer rocketProducer;
  60.  
  61.     @Autowired
  62.     public void setWebSocketServer(ChatMsgService chatMsgService,
  63.                                    RedisTemplate redisTemplate,
  64.                                    RocketProducer rocketProducer) {
  65.         WebSocketMqServer.chatMsgService = chatMsgService;
  66.         WebSocketMqServer.redisTemplate = redisTemplate;
  67.         WebSocketMqServer.rocketProducer = rocketProducer;
  68.     }
  69.  
  70.     @OnOpen
  71.     public void onOpen(Session session, @PathParam(value = "userName") String userName) {
  72.         //1.将用户添加到在线用户列表中
  73.         if (!Constant.SUPER_ADMIN.equals(userName)) {
  74.             redisTemplate.opsForSet().add("online", userName);
  75.         }
  76.  
  77.         //2.保存会话连接
  78.         this.session = session;
  79.         webSockets.add(this);
  80.         sessionPool.put(userName, session);
  81.         Set online = redisTemplate.opsForSet().members("online");
  82.         log.info("【websocket消息】有新的连接，总在线人数为:" + online.size());
  83.  
  84.         //3.创建消息
  85.         WebSocketMqMsg webSocketMqMsg = new WebSocketMqMsg();
  86.         //消息类型
  87.         webSocketMqMsg.setKey(MessageType.MESSAGE_OPEN.getCode());
  88.         //在线人数
  89.         webSocketMqMsg.setOnlineList(online);
  90.         //全部人数
  91.         webSocketMqMsg.setUserList(chatMsgService.getUserList());
  92.  
  93.         //4.消息异步发送到RocketMQ
  94.         rocketProducer.sendAsyncMsg(RocketConstant.ROCKET_TOPIC, RocketConstant.ROCKET_TAG_CHAT, UUID.randomUUID().toString(), JSON.toJSONString(webSocketMqMsg));
  95.     }
  96.  
  97.     @OnClose
  98.     public void onClose(@PathParam(value = "userName") String userName) {
  99.         //1.更新在线用户列表
  100.         redisTemplate.opsForSet().remove("online", userName);
  101.  
  102.         //2.清除会话连接
  103.         webSockets.remove(this);
  104.         sessionPool.remove(userName);
  105.         Set online = redisTemplate.opsForSet().members("online");
  106.         log.info("【websocket消息】连接断开，总在线人数为:" + online.size());
  107.  
  108.         //3.创建消息
  109.         WebSocketMqMsg webSocketMqMsg = new WebSocketMqMsg();
  110.         webSocketMqMsg.setKey(MessageType.MESSAGE_CLOSE.getCode());
  111.         webSocketMqMsg.setOnlineList(online);
  112.         webSocketMqMsg.setUserList(chatMsgService.getUserList());
  113.  
  114.         //4.消息异步发送到RocketMQ
  115.         rocketProducer.sendAsyncMsg(RocketConstant.ROCKET_TOPIC, RocketConstant.ROCKET_TAG_CHAT, UUID.randomUUID().toString(), JSON.toJSONString(webSocketMqMsg));
  116.     }
  117.  
  118.     @OnMessage
  119.     public void onMessage(String message) {
  120.         //1.持久化消息内容
  121.         ChatMsg chatMsg = JSON.parseObject(message, ChatMsg.class);
  122.         chatMsgService.save(chatMsg);
  123.  
  124.         //2.创建消息
  125.         WebSocketMqMsg webSocketMqMsg = new WebSocketMqMsg();
  126.         webSocketMqMsg.setKey(MessageType.MESSAGE_SEND.getCode());
  127.         webSocketMqMsg.setData(chatMsg);
  128.  
  129.         //3.消息异步发送到RocketMQ
  130.         rocketProducer.sendAsyncMsg(RocketConstant.ROCKET_TOPIC, RocketConstant.ROCKET_TAG_CHAT, UUID.randomUUID().toString(), JSON.toJSONString(webSocketMqMsg));
  131.     }
  132.  
  133.     /**
  134.      * 广播消息
  135.      */
  136.     public void sendAllMessage(String message) {
  137.         for (WebSocketMqServer webSocket : webSockets) {
  138.             log.info("【websocket消息】广播消息:" + message);
  139.             try {
  140.                 Session session = webSocket.session;
  141.                 if (session != null && session.isOpen()) {
  142.                     webSocket.session.getAsyncRemote().sendText(message);
  143.                 }
  144.             } catch (Exception e) {
  145.                 e.printStackTrace();
  146.             }
  147.         }
  148.     }
  149.  
  150.     /**
  151.      * 单点消息
  152.      *
  153.      * @param userName
  154.      * @param message
  155.      */
  156.     public void sendOneMessage(String userName, String message) {
  157.         log.info("【websocket消息】单点消息:" + message);
  158.         Session session = sessionPool.get(userName);
  159.         if (session != null && session.isOpen()) {
  160.             try {
  161.                 session.getAsyncRemote().sendText(message);
  162.             } catch (Exception e) {
  163.                 e.printStackTrace();
  164.             }
  165.         }
  166.     }
  167.  
  168.  
  169. }
 
 
 
 
ps:这里我们将会话的消息先推送给消息中间键RocketMQ,然后将消息通过广播的形式分发给每一台服务器去消费，如何能消费成功 ，就将消息推送给对应的客户端
 

 

  • 常量定义
 

 
 
  1.  
  2. package com.yundi.atp.platform.common;
  3.  
  4.  
  5. public class Constant {
  6.     /**
  7.      * zookeeper分布式锁根路径
  8.      */
  9.     public final static String LOCK_ROOT_PATH = "/zookeeper/lock/";
  10.  
  11.     /**
  12.      * websocket协议
  13.      */
  14.     public final static String WEBSOCKET_PROTOCOL = "ws://";
  15.  
  16.     /**
  17.      * 单机版聊天室
  18.      */
  19.     public final static String WEBSOCKET_SINGLE_URL = "/websocket/chat/";
  20.  
  21.  
  22.     /**
  23.      * 微服务版聊天室
  24.      */
  25.     public final static String WEBSOCKET_MQ_URL = "/websocket/mq/chat/";
  26.  
  27.     /**
  28.      * 超级管理员
  29.      */
  30.     public final static String SUPER_ADMIN = "super_admin";
  31. }
 
 
 

 

  • 自定义消息类型：根据不同消息内容处理不同的消息业务逻辑
 

 
 
  1. package com.yundi.atp.platform.enums;
  2.  
  3.  
  4. public enum MessageType {
  5.     MESSAGE_OPEN(1, "开启连接"),
  6.     MESSAGE_CLOSE(2, "断开连接"),
  7.     MESSAGE_SEND(3, "发送消息"),
  8.     MESSAGE_RE_OPEN(4, "异地登录下线通知");
  9.  
  10.     private Integer code;
  11.  
  12.     private String msg;
  13.  
  14.     MessageType(Integer code, String msg) {
  15.         this.code = code;
  16.         this.msg = msg;
  17.     }
  18.  
  19.     public Integer getCode() {
  20.         return code;
  21.     }
  22.  
  23.     public String getMsg() {
  24.         return msg;
  25.     }
  26. }
 
 
​​​​​​​
 

 

  • RocketMQ消息发送的工具类 
 

 
 
  1.  
  2. package com.yundi.atp.platform.rocketmq;
  3.  
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.apache.rocketmq.client.producer.*;
  6. import org.apache.rocketmq.spring.core.RocketMQTemplate;
  7. import org.apache.rocketmq.spring.support.RocketMQHeaders;
  8. import org.springframework.beans.factory.annotation.Autowired;
  9. import org.springframework.messaging.Message;
  10. import org.springframework.messaging.support.MessageBuilder;
  11. import org.springframework.stereotype.Component;
  12.  
  13.  
  14. @Component
  15. @Slf4j
  16. public class RocketProducer {
  17.     @Autowired
  18.     private RocketMQTemplate rocketMQTemplate;
  19.  
  20.     /**
  21.      * 发送同步消息：消息响应后发送下一条消息
  22.      *
  23.      * @param topic 消息主题
  24.      * @param tag   消息tag
  25.      * @param key   业务号
  26.      * @param data  消息内容
  27.      */
  28.     public void sendSyncMsg(String topic, String tag, String key, String data) {
  29.         //消息
  30.         Message message = MessageBuilder.withPayload(data).setHeader(RocketMQHeaders.KEYS, key).build();
  31.         //主题
  32.         String destination = topic + ":" + tag;
  33.         SendResult sendResult = rocketMQTemplate.syncSend(destination, message);
  34.         log.info("【RocketMQ】发送同步消息：{}", sendResult);
  35.     }
  36.  
  37.     /**
  38.      * 发送异步消息：异步回调通知消息发送的状况
  39.      *
  40.      * @param topic 消息主题
  41.      * @param tag   消息tag
  42.      * @param key   业务号
  43.      * @param data  消息内容
  44.      */
  45.     public void sendAsyncMsg(String topic, String tag, String key, String data) {
  46.         //消息
  47.         Message message = MessageBuilder.withPayload(data).setHeader(RocketMQHeaders.KEYS, key).build();
  48.         //主题
  49.         String destination = topic + ":" + tag;
  50.         rocketMQTemplate.asyncSend(destination, message, new SendCallback() {
  51.             @Override
  52.             public void onSuccess(SendResult sendResult) {
  53.                 log.info("【RocketMQ】发送异步消息：{}", sendResult);
  54.             }
  55.  
  56.             @Override
  57.             public void onException(Throwable e) {
  58.                 log.info("【RocketMQ】发送异步消息异常：{}", e.getMessage());
  59.             }
  60.         });
  61.     }
  62.  
  63.  
  64.     /**
  65.      * 发送单向消息：消息发送后无响应，可靠性差，效率高
  66.      *
  67.      * @param topic 消息主题
  68.      * @param tag   消息tag
  69.      * @param key   业务号
  70.      * @param data  消息内容
  71.      */
  72.     public void sendOneWayMsg(String topic, String tag, String key, String data) {
  73.         //消息
  74.         Message message = MessageBuilder.withPayload(data).setHeader(RocketMQHeaders.KEYS, key).build();
  75.         //主题
  76.         String destination = topic + ":" + tag;
  77.         rocketMQTemplate.sendOneWay(destination, message);
  78.     }
  79.  
  80.  
  81.     /**
  82.      * 同步延迟消息
  83.      *
  84.      * @param topic      主题
  85.      * @param tag        标签
  86.      * @param key        业务号
  87.      * @param data       消息体
  88.      * @param timeout    发送消息的过期时间
  89.      * @param delayLevel 延迟等级-----固定等级：1到18分别对应1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
  90.      */
  91.     public void sendSyncDelayMsg(String topic, String tag, String key, String data, long timeout, int delayLevel) {
  92.         //消息
  93.         Message message = MessageBuilder.withPayload(data).setHeader(RocketMQHeaders.KEYS, key).build();
  94.         //主题
  95.         String destination = topic + ":" + tag;
  96.         SendResult sendResult = rocketMQTemplate.syncSend(destination, message, timeout, delayLevel);
  97.         log.info("【RocketMQ】发送同步延迟消息：{}", sendResult);
  98.     }
  99.  
  100.  
  101.     /**
  102.      * 异步延迟消息
  103.      *
  104.      * @param topic      主题
  105.      * @param tag        标签
  106.      * @param key        业务号
  107.      * @param data       消息体
  108.      * @param timeout    发送消息的过期时间
  109.      * @param delayLevel 延迟等级-----固定等级：1到18分别对应1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
  110.      */
  111.     public void sendAsyncDelayMsg(String topic, String tag, String key, String data, long timeout, int delayLevel) {
  112.         //消息
  113.         Message message = MessageBuilder.withPayload(data).setHeader(RocketMQHeaders.KEYS, key).build();
  114.         //主题
  115.         String destination = topic + ":" + tag;
  116.         rocketMQTemplate.asyncSend(destination, message, new SendCallback() {
  117.             @Override
  118.             public void onSuccess(SendResult sendResult) {
  119.                 log.info("【RocketMQ】发送异步延迟消息：{}", sendResult);
  120.             }
  121.  
  122.             @Override
  123.             public void onException(Throwable e) {
  124.                 log.info("【RocketMQ】发送异步延迟消息异常：{}", e.getMessage());
  125.             }
  126.         }, timeout, delayLevel);
  127.     }
  128.  
  129.  
  130.     /**
  131.      * 同步顺序消息
  132.      *
  133.      * @param topic 主题
  134.      * @param tag   标签
  135.      * @param key   业务号
  136.      * @param data  消息体
  137.      */
  138.     public void sendSyncOrderlyMsg(String topic, String tag, String key, String data) {
  139.         //消息
  140.         Message message = MessageBuilder.withPayload(data).setHeader(RocketMQHeaders.KEYS, key).build();
  141.         //主题
  142.         String destination = topic + ":" + tag;
  143.         SendResult sendResult = rocketMQTemplate.syncSendOrderly(destination, message, key);
  144.         log.info("【RocketMQ】发送同步顺序消息：{}", sendResult);
  145.     }
  146.  
  147.  
  148.     /**
  149.      * 异步顺序消息
  150.      *
  151.      * @param topic 主题
  152.      * @param tag   标签
  153.      * @param key   业务号
  154.      * @param data  消息体
  155.      */
  156.     public void sendAsyncOrderlyMsg(String topic, String tag, String key, String data) {
  157.         //消息
  158.         Message message = MessageBuilder.withPayload(data).setHeader(RocketMQHeaders.KEYS, key).build();
  159.         //主题
  160.         String destination = topic + ":" + tag;
  161.         rocketMQTemplate.asyncSendOrderly(destination, message, key, new SendCallback() {
  162.             @Override
  163.             public void onSuccess(SendResult sendResult) {
  164.                 log.info("【RocketMQ】发送异步顺序消息：{}", sendResult);
  165.             }
  166.  
  167.             @Override
  168.             public void onException(Throwable e) {
  169.                 log.info("【RocketMQ】发送异步顺序消息异常：{}", e.getMessage());
  170.             }
  171.         });
  172.     }
  173.  
  174.  
  175.     /**
  176.      * 分布式事务消息
  177.      *
  178.      * @param topic 主题
  179.      * @param tag   标签
  180.      * @param key   业务号
  181.      * @param data  消息体
  182.      */
  183.     public void sendTransactionMsg(String topic, String tag, String key, String data) {
  184.         //消息
  185.         Message message = MessageBuilder.withPayload(data)
  186.                 .setHeader(RocketMQHeaders.KEYS, key)
  187.                 .setHeader(RocketMQHeaders.TRANSACTION_ID, key)
  188.                 .build();
  189.         //主题
  190.         String destination = topic + ":" + tag;
  191.         TransactionSendResult transactionSendResult = rocketMQTemplate.sendMessageInTransaction(destination, message, null);
  192.         if (transactionSendResult.getLocalTransactionState().equals(LocalTransactionState.COMMIT_MESSAGE) &&
  193.                 transactionSendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
  194.             log.info("分布式事物消息发送成功");
  195.  
  196.         }
  197.         log.info("分布式事物消息发送结果:{}", transactionSendResult);
  198.     }
  199.  
  200. }
 
 
 

 

  • websocket服务的连接地址获取及历史消息获取 
 

 
 
  1. package com.yundi.atp.platform.module.test.controller;
  2.  
  3.  
  4. import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
  5. import com.yundi.atp.platform.common.Result;
  6. import com.yundi.atp.platform.module.test.entity.ChatMsg;
  7. import com.yundi.atp.platform.module.test.service.ChatMsgService;
  8. import io.swagger.annotations.Api;
  9. import io.swagger.annotations.ApiOperation;
  10. import org.springframework.beans.factory.annotation.Autowired;
  11. import org.springframework.web.bind.annotation.GetMapping;
  12. import org.springframework.web.bind.annotation.PathVariable;
  13. import org.springframework.web.bind.annotation.RequestMapping;
  14. import org.springframework.web.bind.annotation.RestController;
  15.  
  16. import javax.servlet.http.HttpServletRequest;
  17. import java.net.InetAddress;
  18. import java.net.UnknownHostException;
  19. import java.util.Comparator;
  20. import java.util.List;
  21. import java.util.stream.Collectors;
  22.  
  23. @Api(tags = "聊天室接口-mq版")
  24. @RestController
  25. @RequestMapping("/test/mq/chatMsg")
  26. public class ChatMsgMqController {
  27.     @Autowired
  28.     private ChatMsgService chatMsgService;
  29.  
  30.     @ApiOperation(value = "获取聊天室地址")
  31.     @GetMapping(value = "/getWebSocketAddress/{username}")
  32.     public Result getWebSocketAddress(HttpServletRequest request, @PathVariable(value = "username") String username) throws UnknownHostException {
  33.         String address = "ws://" + InetAddress.getLocalHost().getHostAddress() + ":" + request.getServerPort() + request.getContextPath() + "/websocket/mq/chat/" + username;
  34.         return Result.success(address);
  35.     }
  36.  
  37.     @ApiOperation(value = "获取历史聊天记录")
  38.     @GetMapping(value = "/getHistoryChat/{username}")
  39.     public Result getWebSocketAddress(@PathVariable(value = "username") String username) {
  40.         List<ChatMsg> list = chatMsgService.list(new QueryWrapper<ChatMsg>()
  41.                 .and(wrapper -> wrapper.eq("sender", username).or().eq("receiver", username))
  42.                 .orderByDesc("create_time"));
  43.         List<ChatMsg> collect = list.stream().sorted(Comparator.comparing(ChatMsg::getCreateTime)).collect(Collectors.toList());
  44.         return Result.success(collect);
  45.     }
  46.  
  47.     @ApiOperation(value = "获取用户列表")
  48.     @GetMapping(value = "/getUserList")
  49.     public Result getUserList() {
  50.         List<String> userList = chatMsgService.getUserList();
  51.         return Result.success(userList);
  52.     }
  53.  
  54. }
  55.  
 
 
​​​​​​​
 

 

  • 消息的广播分发
 

 
 
  1. package com.yundi.atp.platform.websocket;
  2.  
  3. import com.alibaba.fastjson.JSON;
  4. import com.yundi.atp.platform.common.Constant;
  5. import com.yundi.atp.platform.enums.MessageType;
  6. import com.yundi.atp.platform.module.test.entity.ChatMsg;
  7. import com.yundi.atp.platform.rocketmq.RocketConstant;
  8. import lombok.extern.slf4j.Slf4j;
  9. import org.apache.rocketmq.spring.annotation.MessageModel;
  10. import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
  11. import org.apache.rocketmq.spring.core.RocketMQListener;
  12. import org.springframework.beans.factory.annotation.Autowired;
  13. import org.springframework.stereotype.Component;
  14.  
  15.  
  16. @Slf4j
  17. @Component
  18. @RocketMQMessageListener(consumerGroup = RocketConstant.ROCKET_CONSUMER_CHAT_GROUP,
  19.         topic = RocketConstant.ROCKET_TOPIC,
  20.         selectorExpression = RocketConstant.ROCKET_TAG_CHAT,
  21.         namespace = RocketConstant.ROCKET_NAMESPACE,
  22.         messageModel = MessageModel.BROADCASTING)
  23. public class WebSocketMqConsumer implements RocketMQListener<String> {
  24.     @Autowired
  25.     WebSocketMqServer webSocketMqServer;
  26.  
  27.     @Override
  28.     public void onMessage(String message) {
  29.         log.info("聊天室消息：{}", message);
  30.         //1.解析消息
  31.         WebSocketMqMsg webSocketMqMsg = JSON.parseObject(message, WebSocketMqMsg.class);
  32.  
  33.         //2.根据消息类型解析消息
  34.  
  35.         // 建立连接消息
  36.         if (webSocketMqMsg.getKey().equals(MessageType.MESSAGE_OPEN.getCode())) {
  37.             webSocketMqServer.sendOneMessage(Constant.SUPER_ADMIN, message);
  38.         }
  39.  
  40.         // 关闭连接消息
  41.         if (webSocketMqMsg.getKey().equals(MessageType.MESSAGE_CLOSE.getCode())) {
  42.             webSocketMqServer.sendOneMessage(Constant.SUPER_ADMIN, message);
  43.         }
  44.  
  45.         // 发送消息
  46.         if (webSocketMqMsg.getKey().equals(MessageType.MESSAGE_SEND.getCode())) {
  47.             ChatMsg data = webSocketMqMsg.getData();
  48.             webSocketMqServer.sendOneMessage(data.getSender(), message);
  49.             webSocketMqServer.sendOneMessage(data.getReceiver(), message);
  50.         }
  51.     }
  52. }
 
 
 

 

  • 消息主题定义
 

 
 
  1. package com.yundi.atp.platform.rocketmq;
  2.  
  3.  
  4. public class RocketConstant {
  5.     /**
  6.      * 消费者组
  7.      */
  8.     public final static String ROCKET_CONSUMER_GROUP = "atp-consumer";
  9.     /**
  10.      * 聊天室消费者组
  11.      */
  12.     public final static String ROCKET_CONSUMER_CHAT_GROUP = "atp-chat-consumer";
  13.     /**
  14.      * 主题
  15.      */
  16.     public final static String ROCKET_TOPIC = "atp";
  17.     /**
  18.      * tag
  19.      */
  20.     public final static String ROCKET_TAG = "app";
  21.     /**
  22.      * 聊天室tag
  23.      */
  24.     public final static String ROCKET_TAG_CHAT = "chat";
  25.     /**
  26.      * 名称空间
  27.      */
  28.     public final static String ROCKET_NAMESPACE = "atp";
  29.  
  30. }
 
 
 

 

  • 客户端代码
 

 
 
  1. <template>
  2.   <div class="container">
  3.     <el-card class="box-card">
  4.       <div slot="header">
  5.         <el-row type="flex">
  6.           <el-col :span="1" style="margin: 15px 10px;">
  7.             <img alt="ATP客服" src="@/assets/logo.png" style="width:40px;height:40px;"/>
  8.           </el-col>
  9.           <el-col :span="3" style="line-height: 74px;margin-left: 10px;">
  10.             <span style="display: inline-block;color: white;">ATP客服</span>
  11.           </el-col>
  12.           <el-col :span="20" v-if="username==='super_admin'">
  13.             <h5 style="color: #83ccd2;padding: 0;text-align: right;margin: 50px 20px 0 0;">当前在线人数：{{ online }}</h5>
  14.           </el-col>
  15.           <el-col :span="20" v-else>
  16.             <h5 style="color: #83ccd2;padding: 0 0 2px 0;text-align: right;margin: 50px 20px 0 0;font-size: 18px;">
  17.               {{ username }}</h5>
  18.           </el-col>
  19.         </el-row>
  20.       </div>
  21.       <div class="content" ref="content">
  22.         <el-row type="flex">
  23.           <el-col :span="6" style="background: #eee;min-height: 600px;" v-if="username==='super_admin'">
  24.             <el-tabs v-model="activeName" @tab-click="handleClick" style="width: 190px;margin: 0 2px;">
  25.               <el-tab-pane label="在线用户" name="online">
  26.                 <div v-for="item in friend" :key="item" @click="switchUser(item)" :class="item===active?'mark':''">
  27.                   <el-badge :is-dot=msgNotify.includes(item) class="item" type="success">
  28.                     <li style="list-style-type:none;padding: 5px 8px;cursor: pointer;"
  29.                         class="active">
  30.                       {{ item }}
  31.                     </li>
  32.                   </el-badge>
  33.                   <el-divider></el-divider>
  34.                 </div>
  35.               </el-tab-pane>
  36.               <el-tab-pane label="全部用户" name="all">
  37.                 <div v-for="item in userList" :key="item" @click="switchUser(item)" :class="item===active?'mark':''">
  38.                   <el-badge :is-dot=msgNotify.includes(item) class="item" type="success">
  39.                     <li style="list-style-type:none;padding: 5px 8px;cursor: pointer;"
  40.                         :class="friend.includes(item)?'active':''">
  41.                       {{ item }}
  42.                     </li>
  43.                   </el-badge>
  44.                   <el-divider></el-divider>
  45.                 </div>
  46.               </el-tab-pane>
  47.             </el-tabs>
  48.           </el-col>
  49.           <el-col :span="18" v-if="username==='super_admin'">
  50.             <div v-for="item in chatMsgList">
  51.               <el-row type="flex" style="margin-bottom: 20px;" v-if="username===item.sender">
  52.                 <el-col :span="2">
  53.                   <img alt="ATP客服" src="@/assets/logo.png" style="width:30px;height:30px;margin: 10px 0px 0px 20px;"/>
  54.                 </el-col>
  55.                 <el-col :span="22">
  56.                   <el-row type="flex" style="margin-top: 10px;margin-left: 5px;opacity: 0.2;">
  57.                     <el-col :span="7"><span style="padding-left: 20px;">{{ item.sender }}</span></el-col>
  58.                     <el-col :span="7"><span>{{ item.createTime | dataFormat('yyyy-MM-dd HH:mm') }}</span></el-col>
  59.                   </el-row>
  60.                   <el-row>
  61.                     <el-col :span="14" style="margin-left: 8px;margin-top: 5px;">
  62.                       <el-card style="padding: 8px 5px;">
  63.                         {{ item.msg }}
  64.                       </el-card>
  65.                     </el-col>
  66.                   </el-row>
  67.                 </el-col>
  68.               </el-row>
  69.               <el-row type="flex" style="margin-bottom: 20px;" v-else justify="end">
  70.                 <el-col :span="22">
  71.                   <el-row type="flex" style="margin-top: 10px;margin-right: 5px;opacity: 0.2;" justify="end">
  72.                     <el-col :span="6"><span>{{ item.sender }}</span></el-col>
  73.                     <el-col :span="7"><span>{{ item.createTime | dataFormat('yyyy-MM-dd HH:mm') }}</span></el-col>
  74.                   </el-row>
  75.                   <el-row type="flex" justify="end" style="margin-right: 8px;margin-top: 5px;">
  76.                     <el-col :span="14" style="margin-right: 8px;">
  77.                       <el-card style="padding: 8px 5px;">
  78.                         {{ item.msg }}
  79.                       </el-card>
  80.                     </el-col>
  81.                   </el-row>
  82.                 </el-col>
  83.                 <el-col :span="2">
  84.                   <el-avatar shape="square" size="medium" style="float: right;margin: 10px 20px 0px 0px;">客户</el-avatar>
  85.                 </el-col>
  86.               </el-row>
  87.             </div>
  88.           </el-col>
  89.           <el-col :span="24" v-else>
  90.             <div v-for="item in chatMsgList">
  91.               <el-row type="flex" style="margin-bottom: 20px;" v-if="username===item.sender">
  92.                 <el-col :span="2">
  93.                   <el-avatar shape="square" size="medium" style="float: right;margin: 10px 20px 0px 0px;">客户</el-avatar>
  94.                 </el-col>
  95.                 <el-col :span="22">
  96.                   <el-row type="flex" style="margin-top: 10px;opacity: 0.2;margin-left: 20px;">
  97.                     <el-col :span="7"><span style="padding-left: 5px;">{{ item.sender }}</span></el-col>
  98.                     <el-col :span="7"><span>{{ item.createTime | dataFormat('yyyy-MM-dd HH:mm') }}</span></el-col>
  99.                   </el-row>
  100.                   <el-row>
  101.                     <el-col :span="14">
  102.                       <el-card style="padding: 8px 5px;">
  103.                         {{ item.msg }}
  104.                       </el-card>
  105.                     </el-col>
  106.                   </el-row>
  107.                 </el-col>
  108.               </el-row>
  109.               <el-row type="flex" style="margin-bottom: 20px;" v-else justify="end">
  110.                 <el-col :span="22">
  111.                   <el-row type="flex" style="margin-top: 10px;margin-right: 5px;opacity: 0.2;" justify="end">
  112.                     <el-col :span="6"><span>{{ item.sender }}</span></el-col>
  113.                     <el-col :span="7"><span>{{ item.createTime | dataFormat('yyyy-MM-dd HH:mm') }}</span></el-col>
  114.                   </el-row>
  115.                   <el-row type="flex" justify="end" style="margin-top: 5px;">
  116.                     <el-col :span="14">
  117.                       <el-card style="padding: 8px 5px;">
  118.                         {{ item.msg }}
  119.                       </el-card>
  120.                     </el-col>
  121.                   </el-row>
  122.                 </el-col>
  123.                 <el-col :span="2">
  124.                   <img alt="ATP客服" src="@/assets/logo.png" style="width:30px;height:30px;margin: 10px 0px 0px 20px;"/>
  125.                 </el-col>
  126.               </el-row>
  127.             </div>
  128.           </el-col>
  129.         </el-row>
  130.       </div>
  131.       <div class="operate" v-if="username==='super_admin'">
  132.         <el-input
  133.             type="textarea"
  134.             :autosize="{ minRows: 3, maxRows: 3}"
  135.             placeholder="您好！这里是ATP客服部，我是客服小美，很高兴为您服务！"
  136.             v-model="msg">
  137.         </el-input>
  138.         <el-button type="success" size="mini" style="float: right;margin-top: 5px;" @click="sendMsg"
  139.                    :disabled="!(msg && active)">
  140.           发送
  141.         </el-button>
  142.       </div>
  143.       <div class="operate" v-else>
  144.         <el-input
  145.             type="textarea"
  146.             :autosize="{ minRows: 3, maxRows: 3}"
  147.             placeholder="您好！这里是ATP客服部，我是客服小美，很高兴为您服务！"
  148.             v-model="msg">
  149.         </el-input>
  150.         <el-button type="success" size="mini" style="float: right;margin-top: 5px;" @click="sendMsg" :disabled="!msg">
  151.           发送
  152.         </el-button>
  153.       </div>
  154.     </el-card>
  155.   </div>
  156. </template>
  157.  
  158. <script>
  159. export default {
  160.   name: "ClientMqChat",
  161.   data() {
  162.     return {
  163.       msg: '',
  164.       chatMsgList: [],
  165.       username: sessionStorage.getItem("username"),
  166.       friend: [],
  167.       online: 0,
  168.       active: '',
  169.       receiver: 'super_admin',
  170.       userList: [],
  171.       activeName: 'online',
  172.       msgNotify:[],
  173.     }
  174.   },
  175.   created() {
  176.     this.getWebSocketAddress();
  177.   },
  178.   methods: {
  179.     //tab切换
  180.     handleClick(tab, event) {
  181.       const _this = this;
  182.       if (tab.name === 'online') {
  183.         if (!_this.active) {
  184.           if (_this.online > 0) {
  185.             _this.active = _this.friend[0];
  186.             _this.activeName = 'online';
  187.             _this.receiver = _this.active;
  188.             _this.getHistoryChat(_this.receiver);
  189.           } else {
  190.             if (_this.userList.length > 0) {
  191.               _this.active = _this.userList[0];
  192.               _this.activeName = 'all';
  193.               _this.receiver = _this.active;
  194.               _this.getHistoryChat(_this.receiver);
  195.             }
  196.           }
  197.         }
  198.       }
  199.       if (tab.name === 'all') {
  200.         if (!_this.active) {
  201.           if (_this.online > 0) {
  202.             _this.active = _this.friend[0];
  203.             _this.activeName = 'online';
  204.             _this.receiver = _this.active;
  205.             _this.getHistoryChat(_this.receiver);
  206.           } else {
  207.             if (_this.userList.length > 0) {
  208.               _this.active = _this.userList[0];
  209.               _this.activeName = 'all';
  210.               _this.receiver = _this.active;
  211.               _this.getHistoryChat(_this.receiver);
  212.             }
  213.           }
  214.         }
  215.       }
  216.     },
  217.     //切换用户
  218.     switchUser(data) {
  219.       if (this.active === data) {
  220.         return;
  221.       }
  222.       this.active = data;
  223.       this.receiver = data;
  224.       //获取历史聊天记录
  225.       this.getHistoryChat(this.receiver);
  226.       this.msgNotify = this.msgNotify.filter(item => item != this.active);
  227.     },
  228.     //获取历史聊天记录
  229.     getHistoryChat(data) {
  230.       this.$http.get('/test/mq/chatMsg/getHistoryChat/' + data).then(res => {
  231.         if (res.data.code === 1) {
  232.           this.chatMsgList = res.data.data;
  233.           this.flushScroll();
  234.         } else {
  235.           this.$message.warning(res.data.msg);
  236.         }
  237.       }).catch(error => {
  238.         this.$message.error(error);
  239.       });
  240.     },
  241.     //获取websocket地址
  242.     getWebSocketAddress() {
  243.       this.$http.get('/test/mq/chatMsg/getWebSocketAddress/' + this.username).then(res => {
  244.         if (res.data.code === 1) {
  245.           if ('WebSocket' in window) {
  246.             this.websocket = new WebSocket(res.data.data);
  247.             this.initWebSocket();
  248.             if (this.username != 'super_admin') {
  249.               this.getHistoryChat(this.username);
  250.             }
  251.           } else {
  252.             this.$message.warning('当前浏览器不支持websocket创建！');
  253.           }
  254.         } else {
  255.           this.$message.warning(res.data.msg);
  256.         }
  257.       }).catch(error => {
  258.         this.$message.error(error);
  259.       });
  260.     },
  261.     //初始化websocket
  262.     initWebSocket() {
  263.       const _this = this;
  264.       _this.websocket.onerror = function (event) {
  265.         _this.$message.error('服务端连接错误！');
  266.       }
  267.       _this.websocket.onopen = function (event) {
  268.         _this.$message.success("连接成功！");
  269.       }
  270.       _this.websocket.onmessage = function (event) {
  271.         let res = JSON.parse(event.data);
  272.         if (res.key === 1) {
  273.           _this.userList = res.userList;
  274.           _this.friend = res.onlineList;
  275.           _this.online = _this.friend.length;
  276.           if (!_this.active) {
  277.             if (_this.online > 0) {
  278.               _this.active = _this.friend[0];
  279.               _this.activeName = 'online';
  280.               _this.receiver = _this.active;
  281.               _this.getHistoryChat(_this.receiver);
  282.             } else {
  283.               if (_this.userList.length > 0) {
  284.                 _this.active = _this.userList[0];
  285.                 _this.activeName = 'all';
  286.                 _this.receiver = _this.active;
  287.                 _this.getHistoryChat(_this.receiver);
  288.               }
  289.             }
  290.           }
  291.         }
  292.         if (res.key === 2) {
  293.           _this.userList = res.userList;
  294.           _this.friend = res.onlineList;
  295.           _this.online = _this.friend.length;
  296.           if (!_this.active) {
  297.             if (_this.online > 0) {
  298.               _this.active = _this.friend[0];
  299.               _this.activeName = 'online';
  300.               _this.receiver = _this.active;
  301.               _this.getHistoryChat(_this.receiver);
  302.             } else {
  303.               if (_this.userList.length > 0) {
  304.                 _this.active = _this.userList[0];
  305.                 _this.activeName = 'all';
  306.                 _this.receiver = _this.active;
  307.                 _this.getHistoryChat(_this.receiver);
  308.               }
  309.             }
  310.           }
  311.         }
  312.         if (res.key === 3) {
  313.           if (_this.username === res.data.sender) {
  314.             _this.chatMsgList.push(res.data);
  315.             _this.flushScroll();
  316.           } else {
  317.             if (res.data.sender === 'super_admin') {
  318.               _this.chatMsgList.push(res.data);
  319.               _this.flushScroll();
  320.             } else {
  321.               if (res.data.sender === _this.active) {
  322.                 _this.chatMsgList.push(res.data);
  323.                 _this.flushScroll();
  324.               } else {
  325.                 //发送其它用户处理
  326.                 _this.msgNotify.push(res.data.sender);
  327.               }
  328.             }
  329.           }
  330.         }
  331.       }
  332.       _this.websocket.onclose = function (event) {
  333.         _this.$message.warning('服务端已关闭！');
  334.       }
  335.     },
  336.     //发送消息
  337.     sendMsg() {
  338.       if (this.msg.trim().length === 0) {
  339.         this.$message.warning('不能发送空消息！');
  340.         return;
  341.       }
  342.       let chatMsg = {};
  343.       chatMsg.msg = this.msg;
  344.       chatMsg.sender = this.username;
  345.       chatMsg.createTime = new Date();
  346.       chatMsg.receiver = this.receiver;
  347.       this.websocket.send(JSON.stringify(chatMsg));
  348.       this.msg = '';
  349.       this.flushScroll();
  350.     },
  351.     //刷新滚动条
  352.     flushScroll() {
  353.       let content = this.$refs.content;
  354.       setTimeout(() => {
  355.         content.scrollTop = content.scrollHeight;
  356.       }, 100);
  357.     },
  358.   }
  359. }
  360. </script>
  361.  
  362. <style scoped lang="scss">
  363. .container {
  364.   padding-top: 50px;
  365.  
  366.   .box-card {
  367.     margin: auto;
  368.     width: 800px;
  369.     height: 800px;
  370.     max-height: 900px;
  371.  
  372.     ::v-deep .el-card__header {
  373.       background: #867ba9 !important;
  374.       border-bottom: none;
  375.       padding: 0;
  376.     }
  377.  
  378.     ::v-deep .el-card__body {
  379.       padding: 0px !important;
  380.       position: relative;
  381.  
  382.       .content {
  383.         height: 600px;
  384.         background: #ddd;
  385.         overflow-y: auto;
  386.  
  387.         .el-divider--horizontal {
  388.           margin: 0;
  389.         }
  390.  
  391.         .active {
  392.           color: #0080ff;
  393.         }
  394.  
  395.         .mark {
  396.           background: #deb068;
  397.         }
  398.  
  399.         .item {
  400.           margin-top: 10px;
  401.           margin-right: 10px;
  402.         }
  403.       }
  404.  
  405.       .operate {
  406.         padding: 5px 15px;
  407.       }
  408.     }
  409.   }
  410. }
  411. </style>
  412.  
 
 
 

 

  • ​​​​​​​ 启动前后端项目，分别使用客服账号和客户账号登录聊天室
 

 

  • 聊天消息
 

 

 
 

 
 

结语
 

至此，关于实现微服务的websocket聊天室到这里就结束了，下期见。。。

                

        
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/blog/article/detail/49691
推荐阅读
相关标签
Copyright ©  2003-2013 www.wpsshop.cn  版权所有，并保留所有权利。

  
闽ICP备14008679号