关键词：
前言
在前面的博客内容中我们介绍了如何使用websocket实现一个网页版的在线客服聊天室，众所周知，由于websocket是一个长连接，要和服务端保持会话连接，所以其本身并不适用于微服务环境，在微服务环境中，有可能A、B俩个客户端连接到不同的服务A、B中，这样就没法保证A、B俩个客户端完成聊天的功能，因为会话不在同一台服务器上，A、B无法感知到对方发送的消息，为了解决websocket单机的这个痛点，我们引入消息中间键RocketMQ的广播机制，实现消息的转发，从而实现微服务版的websocke聊天室功能。其架构如下：
本节内容使用的主要技术包含springboot、redis、rocketmq、vue等，关于中间键的搭建本节内容不在展开，请关注作者的往期博客内容。
正文
- 引入websocket、redis和rocketmq的pom依赖
①核心pom依赖
<!-- rocketmq--> <dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-spring-boot-starter</artifactId> <version>2.2.2</version> </dependency> <!-- websocket--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-websocket</artifactId> </dependency> <!-- redis--> <dependency> <groupId>org.springframework.session</groupId> <artifactId>spring-session-data-redis</artifactId> <version>2.4.3</version> </dependency>
PS：可以按需引入自己需要的依赖，作者这里只列出核心的pom依赖
-
配置application.yml
①配置文件
server: port: 8888 spring: #数据源配置 datasource: dynamic: primary: master #设置默认的数据源或者数据源组,默认值即为master strict: false #设置严格模式,默认false不启动. 启动后在未匹配到指定数据源时候会抛出异常,不启动则使用默认数据源. datasource: master: url: jdbc:mysql://192.168.56.10:3306/atp username: root password: root driver-class-name: com.mysql.cj.jdbc.Driver # 3.2.0开始支持SPI可省略此配置 profiles: active: dev servlet: multipart: max-file-size: 52428800 max-request-size: 52428800 #redis配置 redis: #redisson配置 redisson: file: classpath:redisson.yaml #默认数据分区 database: 0 #redis集群节点配置 cluster: nodes: - 192.168.56.10:6379 - 192.168.56.10:6380 - 192.168.56.10.6381 max-redirects: 3 #超时时间 timeout: 10000 #哨兵节点配置 sentinel: master: mymaster nodes: - "192.168.56.10:26379" - "192.168.56.10:26380" - "192.168.56.10:26381" #redis密码 password: root #redis 客户端工具 lettuce: pool: # 连接池最大连接数(使用负值表示没有限制) 默认为8 max-active: 8 # 连接池中的最小空闲连接 默认为 0 min-idle: 1 # 连接池最大阻塞等待时间(使用负值表示没有限制) 默认为-1 max-wait: 1000 # 连接池中的最大空闲连接 默认为8 max-idle: 8 session: store-type: redis redis: flush-mode: on_save namespace: spring:session:atp thymeleaf: cache: false #mybatisplus配置 mybatis-plus: mapper-locations: classpath*:/mapper/*/*Mapper.xml type-aliases-package: com.yundi.atp.platform.module.*.entity configuration: map-underscore-to-camel-case: true global-config: db-config: id-type: assign_id #rocketmq配置 rocketmq: #注册地址 name-server: 192.168.56.10:9876;192.168.56.10:9877 producer: #生产者组名称 group: atp-producer #命名空间 namespace: atp #异步消息发送失败重试次数，默认是2 retry-times-when-send-async-failed: 2 #发送消息超时时间，默认2000ms send-message-timeout: 2000 #消息的最大长度：默认1024 * 1024 * 4(默认4M) max-message-size: 40000000 #压缩消息阈值，超过4k就压缩 compress-message-body-threshold: 4096 #是否发送失败，重试另外的broker retry-next-server: false #是否启用消息追踪 enable-msg-trace: false #默认追踪的主题 customized-trace-topic: RMQ_SYS_TRACE_TOPIC #消息发送失败重试的次数 retry-times-when-send-failed: 2
-
创建websocket服务配置WebSocketConfig.java
package com.yundi.atp.platform.websocket; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.web.socket.server.standard.ServerEndpointExporter; @Configuration public class WebSocketConfig /** * 注入ServerEndpointExporter， * 这个bean会自动注册使用了@ServerEndpoint注解声明的Websocket endpoint */ @Bean public ServerEndpointExporter serverEndpointExporter() return new ServerEndpointExporter();
-
创建微服务版websocket服务
package com.yundi.atp.platform.websocket; import com.alibaba.fastjson.JSON; import com.yundi.atp.platform.common.Constant; import com.yundi.atp.platform.enums.MessageType; import com.yundi.atp.platform.module.test.entity.ChatMsg; import com.yundi.atp.platform.module.test.service.ChatMsgService; import com.yundi.atp.platform.rocketmq.RocketConstant; import com.yundi.atp.platform.rocketmq.RocketProducer; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.stereotype.Component; import javax.websocket.OnClose; import javax.websocket.OnMessage; import javax.websocket.OnOpen; import javax.websocket.Session; import javax.websocket.server.PathParam; import javax.websocket.server.ServerEndpoint; import java.util.Map; import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArraySet; @Slf4j @Component @ServerEndpoint(Constant.WEBSOCKET_MQ_URL + "userName") public class WebSocketMqServer /** * 会话session */ private Session session; /** * socket连接 */ private static CopyOnWriteArraySet<WebSocketMqServer> webSockets = new CopyOnWriteArraySet<>(); /** * 会话连接池 */ private static Map<String, Session> sessionPool = new ConcurrentHashMap<>(); /** * 消息持久化 */ private static ChatMsgService chatMsgService; /** * redis */ private static RedisTemplate redisTemplate; /** * RocketMQ消息工具类 */ private static RocketProducer rocketProducer; @Autowired public void setWebSocketServer(ChatMsgService chatMsgService, RedisTemplate redisTemplate, RocketProducer rocketProducer) WebSocketMqServer.chatMsgService = chatMsgService; WebSocketMqServer.redisTemplate = redisTemplate; WebSocketMqServer.rocketProducer = rocketProducer; @OnOpen public void onOpen(Session session, @PathParam(value = "userName") String userName) //1.将用户添加到在线用户列表中 if (!Constant.SUPER_ADMIN.equals(userName)) redisTemplate.opsForSet().add("online", userName); //2.保存会话连接 this.session = session; webSockets.add(this); sessionPool.put(userName, session); Set online = redisTemplate.opsForSet().members("online"); log.info("【websocket消息】有新的连接，总在线人数为:" + online.size()); //3.创建消息 WebSocketMqMsg webSocketMqMsg = new WebSocketMqMsg(); //消息类型 webSocketMqMsg.setKey(MessageType.MESSAGE_OPEN.getCode()); //在线人数 webSocketMqMsg.setOnlineList(online); //全部人数 webSocketMqMsg.setUserList(chatMsgService.getUserList()); //4.消息异步发送到RocketMQ rocketProducer.sendAsyncMsg(RocketConstant.ROCKET_TOPIC, RocketConstant.ROCKET_TAG_CHAT, UUID.randomUUID().toString(), JSON.toJSONString(webSocketMqMsg)); @OnClose public void onClose(@PathParam(value = "userName") String userName) //1.更新在线用户列表 redisTemplate.opsForSet().remove("online", userName); //2.清除会话连接 webSockets.remove(this); sessionPool.remove(userName); Set online = redisTemplate.opsForSet().members("online"); log.info("【websocket消息】连接断开，总在线人数为:" + online.size()); //3.创建消息 WebSocketMqMsg webSocketMqMsg = new WebSocketMqMsg(); webSocketMqMsg.setKey(MessageType.MESSAGE_CLOSE.getCode()); webSocketMqMsg.setOnlineList(online); webSocketMqMsg.setUserList(chatMsgService.getUserList()); //4.消息异步发送到RocketMQ rocketProducer.sendAsyncMsg(RocketConstant.ROCKET_TOPIC, RocketConstant.ROCKET_TAG_CHAT, UUID.randomUUID().toString(), JSON.toJSONString(webSocketMqMsg)); @OnMessage public void onMessage(String message) //1.持久化消息内容 ChatMsg chatMsg = JSON.parseObject(message, ChatMsg.class); chatMsgService.save(chatMsg); //2.创建消息 WebSocketMqMsg webSocketMqMsg = new WebSocketMqMsg(); webSocketMqMsg.setKey(MessageType.MESSAGE_SEND.getCode()); webSocketMqMsg.setData(chatMsg); //3.消息异步发送到RocketMQ rocketProducer.sendAsyncMsg(RocketConstant.ROCKET_TOPIC, RocketConstant.ROCKET_TAG_CHAT, UUID.randomUUID().toString(), JSON.toJSONString(webSocketMqMsg)); /** * 广播消息 */ public void sendAllMessage(String message) for (WebSocketMqServer webSocket : webSockets) log.info("【websocket消息】广播消息:" + message); try Session session = webSocket.session; if (session != null && session.isOpen()) webSocket.session.getAsyncRemote().sendText(message); catch (Exception e) e.printStackTrace(); /** * 单点消息 * * @param userName * @param message */ public void sendOneMessage(String userName, String message) log.info("【websocket消息】单点消息:" + message); Session session = sessionPool.get(userName); if (session != null && session.isOpen()) try session.getAsyncRemote().sendText(message); catch (Exception e) e.printStackTrace();
ps:这里我们将会话的消息先推送给消息中间键RocketMQ,然后将消息通过广播的形式分发给每一台服务器去消费，如何能消费成功 ，就将消息推送给对应的客户端
- 常量定义
package com.yundi.atp.platform.common; public class Constant /** * zookeeper分布式锁根路径 */ public final static String LOCK_ROOT_PATH = "/zookeeper/lock/"; /** * websocket协议 */ public final static String WEBSOCKET_PROTOCOL = "ws://"; /** * 单机版聊天室 */ public final static String WEBSOCKET_SINGLE_URL = "/websocket/chat/"; /** * 微服务版聊天室 */ public final static String WEBSOCKET_MQ_URL = "/websocket/mq/chat/"; /** * 超级管理员 */ public final static String SUPER_ADMIN = "super_admin";
- 自定义消息类型：根据不同消息内容处理不同的消息业务逻辑
package com.yundi.atp.platform.enums; public enum MessageType MESSAGE_OPEN(1, "开启连接"), MESSAGE_CLOSE(2, "断开连接"), MESSAGE_SEND(3, "发送消息"), MESSAGE_RE_OPEN(4, "异地登录下线通知"); private Integer code; private String msg; MessageType(Integer code, String msg) this.code = code; this.msg = msg; public Integer getCode() return code; public String getMsg() return msg;
- RocketMQ消息发送的工具类
package com.yundi.atp.platform.rocketmq; import lombok.extern.slf4j.Slf4j; import org.apache.rocketmq.client.producer.*; import org.apache.rocketmq.spring.core.RocketMQTemplate; import org.apache.rocketmq.spring.support.RocketMQHeaders; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.messaging.Message; import org.springframework.messaging.support.MessageBuilder; import org.springframework.stereotype.Component; @Component @Slf4j public class RocketProducer @Autowired private RocketMQTemplate rocketMQTemplate; /** * 发送同步消息：消息响应后发送下一条消息 * * @param topic 消息主题 * @param tag 消息tag * @param key 业务号 * @param data 消息内容 */ public void sendSyncMsg(String topic, String tag, String key, String data) //消息 Message message = MessageBuilder.withPayload(data).setHeader(RocketMQHeaders.KEYS, key).build(); //主题 String destination = topic + ":" + tag; SendResult sendResult = rocketMQTemplate.syncSend(destination, message); log.info("【RocketMQ】发送同步消息：", sendResult); /** * 发送异步消息：异步回调通知消息发送的状况 * * @param topic 消息主题 * @param tag 消息tag * @param key 业务号 * @param data 消息内容 */ public void sendAsyncMsg(String topic, String tag, String key, String data) //消息 Message message = MessageBuilder.withPayload(data).setHeader(RocketMQHeaders.KEYS, key).build(); //主题 String destination = topic + ":" + tag; rocketMQTemplate.asyncSend(destination, message, new SendCallback() @Override public void onSuccess(SendResult sendResult) log.info("【RocketMQ】发送异步消息：", sendResult); @Override public void onException(Throwable e) log.info("【RocketMQ】发送异步消息异常：", e.getMessage()); ); /** * 发送单向消息：消息发送后无响应，可靠性差，效率高 * * @param topic 消息主题 * @param tag 消息tag * @param key 业务号 * @param data 消息内容 */ public void sendOneWayMsg(String topic, String tag, String key, String data) //消息 Message message = MessageBuilder.withPayload(data).setHeader(RocketMQHeaders.KEYS, key).build(); //主题 String destination = topic + ":" + tag; rocketMQTemplate.sendOneWay(destination, message); /** * 同步延迟消息 * * @param topic 主题 * @param tag 标签 * @param key 业务号 * @param data 消息体 * @param timeout 发送消息的过期时间 * @param delayLevel 延迟等级-----固定等级：1到18分别对应1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h */ public void sendSyncDelayMsg(String topic, String tag, String key, String data, long timeout, int delayLevel) //消息 Message message = MessageBuilder.withPayload(data).setHeader(RocketMQHeaders.KEYS, key).build(); //主题 String destination = topic + ":" + tag; SendResult sendResult = rocketMQTemplate.syncSend(destination, message, timeout, delayLevel); log.info("【RocketMQ】发送同步延迟消息：", sendResult); /** * 异步延迟消息 * * @param topic 主题 * @param tag 标签 * @param key 业务号 * @param data 消息体 * @param timeout 发送消息的过期时间 * @param delayLevel 延迟等级-----固定等级：1到18分别对应1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h */ public void sendAsyncDelayMsg(String topic, String tag, String key, String data, long timeout, int delayLevel) //消息 Message message = MessageBuilder.withPayload(data).setHeader(RocketMQHeaders.KEYS, key).build(); //主题 String destination = topic + ":" + tag; rocketMQTemplate.asyncSend(destination, message, new SendCallback() @Override public void onSuccess(SendResult sendResult) log.info("【RocketMQ】发送异步延迟消息：", sendResult); @Override public void onException(Throwable e) log.info("【RocketMQ】发送异步延迟消息异常：", e.getMessage()); , timeout, delayLevel); /** * 同步顺序消息 * * @param topic 主题 * @param tag 标签 * @param key 业务号 * @param data 消息体 */ public void sendSyncOrderlyMsg(String topic, String tag, String key, String data) //消息 Message message = MessageBuilder.withPayload(data).setHeader(RocketMQHeaders.KEYS, key).build(); //主题 String destination = topic + ":" + tag; SendResult sendResult = rocketMQTemplate.syncSendOrderly(destination, message, key); log.info("【RocketMQ】发送同步顺序消息：", sendResult); /** * 异步顺序消息 * * @param topic 主题 * @param tag 标签 * @param key 业务号 * @param data 消息体 */ public void sendAsyncOrderlyMsg(String topic, String tag, String key, String data) //消息 Message message = MessageBuilder.withPayload(data).setHeader(RocketMQHeaders.KEYS, key).build(); //主题 String destination = topic + ":" + tag; rocketMQTemplate.asyncSendOrderly(destination, message, key, new SendCallback() @Override public void onSuccess(SendResult sendResult) log.info("【RocketMQ】发送异步顺序消息：", sendResult); @Override public void onException(Throwable e) log.info("【RocketMQ】发送异步顺序消息异常：", e.getMessage()); ); /** * 分布式事务消息 * * @param topic 主题 * @param tag 标签 * @param key 业务号 * @param data 消息体 */ public void sendTransactionMsg(String topic, String tag, String key, String data) //消息 Message message = MessageBuilder.withPayload(data) .setHeader(RocketMQHeaders.KEYS, key) .setHeader(RocketMQHeaders.TRANSACTION_ID, key) .build(); //主题 String destination = topic + ":" + tag; TransactionSendResult transactionSendResult = rocketMQTemplate.sendMessageInTransaction(destination, message, null); if (transactionSendResult.getLocalTransactionState().equals(LocalTransactionState.COMMIT_MESSAGE) && transactionSendResult.getSendStatus().equals(SendStatus.SEND_OK)) log.info("分布式事物消息发送成功"); log.info("分布式事物消息发送结果:", transactionSendResult);
- websocket服务的连接地址获取及历史消息获取
package com.yundi.atp.platform.module.test.controller; import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper; import com.yundi.atp.platform.common.Result; import com.yundi.atp.platform.module.test.entity.ChatMsg; import com.yundi.atp.platform.module.test.service.ChatMsgService; import io.swagger.annotations.Api; import io.swagger.annotations.ApiOperation; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; import javax.servlet.http.HttpServletRequest; import java.net.InetAddress; import java.net.UnknownHostException; import java.util.Comparator; import java.util.List; import java.util.stream.Collectors; @Api(tags = "聊天室接口-mq版") @RestController @RequestMapping("/test/mq/chatMsg") public class ChatMsgMqController @Autowired private ChatMsgService chatMsgService; @ApiOperation(value = "获取聊天室地址") @GetMapping(value = "/getWebSocketAddress/username") public Result getWebSocketAddress(HttpServletRequest request, @PathVariable(value = "username") String username) throws UnknownHostException String address = "ws://" + InetAddress.getLocalHost().getHostAddress() + ":" + request.getServerPort() + request.getContextPath() + "/websocket/mq/chat/" + username; return Result.success(address); @ApiOperation(value = "获取历史聊天记录") @GetMapping(value = "/getHistoryChat/username") public Result getWebSocketAddress(@PathVariable(value = "username") String username) List<ChatMsg> list = chatMsgService.list(new QueryWrapper<ChatMsg>() .and(wrapper -> wrapper.eq("sender", username).or().eq("receiver", username)) .orderByDesc("create_time")); List<ChatMsg> collect = list.stream().sorted(Comparator.comparing(ChatMsg::getCreateTime)).collect(Collectors.toList()); return Result.success(collect); @ApiOperation(value = "获取用户列表") @GetMapping(value = "/getUserList") public Result getUserList() List<String> userList = chatMsgService.getUserList(); return Result.success(userList);
- 消息的广播分发
package com.yundi.atp.platform.websocket; import com.alibaba.fastjson.JSON; import com.yundi.atp.platform.common.Constant; import com.yundi.atp.platform.enums.MessageType; import com.yundi.atp.platform.module.test.entity.ChatMsg; import com.yundi.atp.platform.rocketmq.RocketConstant; import lombok.extern.slf4j.Slf4j; import org.apache.rocketmq.spring.annotation.MessageModel; import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.core.RocketMQListener; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; @Slf4j @Component @RocketMQMessageListener(consumerGroup = RocketConstant.ROCKET_CONSUMER_CHAT_GROUP, topic = RocketConstant.ROCKET_TOPIC, selectorExpression = RocketConstant.ROCKET_TAG_CHAT, namespace = RocketConstant.ROCKET_NAMESPACE, messageModel = MessageModel.BROADCASTING) public class WebSocketMqConsumer implements RocketMQListener<String> @Autowired WebSocketMqServer webSocketMqServer; @Override public void onMessage(String message) log.info("聊天室消息：", message); //1.解析消息 WebSocketMqMsg webSocketMqMsg = JSON.parseObject(message, WebSocketMqMsg.class); //2.根据消息类型解析消息 // 建立连接消息 if (webSocketMqMsg.getKey().equals(MessageType.MESSAGE_OPEN.getCode())) webSocketMqServer.sendOneMessage(Constant.SUPER_ADMIN, message); // 关闭连接消息 if (webSocketMqMsg.getKey().equals(MessageType.MESSAGE_CLOSE.getCode())) webSocketMqServer.sendOneMessage(Constant.SUPER_ADMIN, message); // 发送消息 if (webSocketMqMsg.getKey().equals(MessageType.MESSAGE_SEND.getCode())) ChatMsg data = webSocketMqMsg.getData(); webSocketMqServer.sendOneMessage(data.getSender(), message); webSocketMqServer.sendOneMessage(data.getReceiver(), message);
- 消息主题定义
package com.yundi.atp.platform.rocketmq; public class RocketConstant /** * 消费者组 */ public final static String ROCKET_CONSUMER_GROUP = "atp-consumer"; /** * 聊天室消费者组 */ public final static String ROCKET_CONSUMER_CHAT_GROUP = "atp-chat-consumer"; /** * 主题 */ public final static String ROCKET_TOPIC = "atp"; /** * tag */ public final static String ROCKET_TAG = "app"; /** * 聊天室tag */ public final static String ROCKET_TAG_CHAT = "chat"; /** * 名称空间 */ public final static String ROCKET_NAMESPACE = "atp";
- 客户端代码
<template> <div class="container"> <el-card class="box-card"> <div slot="header"> <el-row type="flex"> <el-col :span="1" style="margin: 15px 10px;"> <img alt="ATP客服" src="@/assets/logo.png" style="width:40px;height:40px;"/> </el-col> <el-col :span="3" style="line-height: 74px;margin-left: 10px;"> <span style="display: inline-block;color: white;">ATP客服</span> </el-col> <el-col :span="20" v-if="username==='super_admin'"> <h5 style="color: #83ccd2;padding: 0;text-align: right;margin: 50px 20px 0 0;">当前在线人数： online </h5> </el-col> <el-col :span="20" v-else> <h5 style="color: #83ccd2;padding: 0 0 2px 0;text-align: right;margin: 50px 20px 0 0;font-size: 18px;"> username </h5> </el-col> </el-row> </div> <div class="content" ref="content"> <el-row type="flex"> <el-col :span="6" style="background: #eee;min-height: 600px;" v-if="username==='super_admin'"> <el-tabs v-model="activeName" @tab-click="handleClick" style="width: 190px;margin: 0 2px;"> <el-tab-pane label="在线用户" name="online"> <div v-for="item in friend" :key="item" @click="switchUser(item)" :class="item===active?'mark':''"> <el-badge :is-dot=msgNotify.includes(item) class="item" type="success"> <li style="list-style-type:none;padding: 5px 8px;cursor: pointer;" class="active"> item </li> </el-badge> <el-divider></el-divider> </div> </el-tab-pane> <el-tab-pane label="全部用户" name="all"> <div v-for="item in userList" :key="item" @click="switchUser(item)" :class="item===active?'mark':''"> <el-badge :is-dot=msgNotify.includes(item) class="item" type="success"> <li style="list-style-type:none;padding: 5px 8px;cursor: pointer;" :class="friend.includes(item)?'active':''"> item </li> </el-badge> <el-divider></el-divider> </div> </el-tab-pane> </el-tabs> </el-col> <el-col :span="18" v-if="username==='super_admin'"> <div v-for="item in chatMsgList"> <el-row type="flex" style="margin-bottom: 20px;" v-if="username===item.sender"> <el-col :span="2"> <img alt="ATP客服" src="@/assets/logo.png" style="width:30px;height:30px;margin: 10px 0px 0px 20px;"/> </el-col> <el-col :span="22"> <el-row type="flex" style="margin-top: 10px;margin-left: 5px;opacity: 0.2;"> <el-col :span="7"><span style="padding-left: 20px;"> item.sender </span></el-col> <el-col :span="7"><span> item.createTime | dataFormat('yyyy-MM-dd HH:mm') </span></el-col> </el-row> <el-row> <el-col :span="14" style="margin-left: 8px;margin-top: 5px;"> <el-card style="padding: 8px 5px;"> item.msg </el-card> </el-col> </el-row> </el-col> </el-row> <el-row type="flex" style="margin-bottom: 20px;" v-else justify="end"> <el-col :span="22"> <el-row type="flex" style="margin-top: 10px;margin-right: 5px;opacity: 0.2;" justify="end"> <el-col :span="6"><span> item.sender </span></el-col> <el-col :span="7"><span> item.createTime | dataFormat('yyyy-MM-dd HH:mm') </span></el-col> </el-row> <el-row type="flex" justify="end" style="margin-right: 8px;margin-top: 5px;"> <el-col :span="14" style="margin-right: 8px;"> <el-card style="padding: 8px 5px;"> item.msg </el-card> </el-col> </el-row> </el-col> <el-col :span="2"> <el-avatar shape="square" size="medium" style="float: right;margin: 10px 20px 0px 0px;">客户</el-avatar> </el-col> </el-row> </div> </el-col> <el-col :span="24" v-else> <div v-for="item in chatMsgList"> <el-row type="flex" style="margin-bottom: 20px;" v-if="username===item.sender"> <el-col :span="2"> <el-avatar shape="square" size="medium" style="float: right;margin: 10px 20px 0px 0px;">客户</el-avatar> </el-col> <el-col :span="22"> <el-row type="flex" style="margin-top: 10px;opacity: 0.2;margin-left: 20px;"> <el-col :span="7"><span style="padding-left: 5px;"> item.sender </span></el-col> <el-col :span="7"><span> item.createTime | dataFormat('yyyy-MM-dd HH:mm') </span></el-col> </el-row> <el-row> <el-col :span="14"> <el-card style="padding: 8px 5px;"> item.msg </el-card> </el-col> </el-row> </el-col> </el-row> <el-row type="flex" style="margin-bottom: 20px;" v-else justify="end"> <el-col :span="22"> <el-row type="flex" style="margin-top: 10px;margin-right: 5px;opacity: 0.2;" justify="end"> <el-col :span="6"><span> item.sender </span></el-col> <el-col :span="7"><span> item.createTime | dataFormat('yyyy-MM-dd HH:mm') </span></el-col> </el-row> <el-row type="flex" justify="end" style="margin-top: 5px;"> <el-col :span="14"> <el-card style="padding: 8px 5px;"> item.msg </el-card> </el-col> </el-row> </el-col> <el-col :span="2"> <img alt="ATP客服" src="@/assets/logo.png" style="width:30px;height:30px;margin: 10px 0px 0px 20px;"/> </el-col> </el-row> </div> </el-col> </el-row> </div> <div class="operate" v-if="username==='super_admin'"> <el-input type="textarea" :autosize=" minRows: 3, maxRows: 3" placeholder="您好！这里是ATP客服部，我是客服小美，很高兴为您服务！" v-model="msg"> </el-input> <el-button type="success" size="mini" style="float: right;margin-top: 5px;" @click="sendMsg" :disabled="!(msg && active)"> 发送 </el-button> </div> <div class="operate" v-else> <el-input type="textarea" :autosize=" minRows: 3, maxRows: 3" placeholder="您好！这里是ATP客服部，我是客服小美，很高兴为您服务！" v-model="msg"> </el-input> <el-button type="success" size="mini" style="float: right;margin-top: 5px;" @click="sendMsg" :disabled="!msg"> 发送 </el-button> </div> </el-card> </div> </template> <script> export default name: "ClientMqChat", data() return msg: '', chatMsgList: [], username: sessionStorage.getItem("username"), friend: [], online: 0, active: '', receiver: 'super_admin', userList: [], activeName: 'online', msgNotify:[], , created() this.getWebSocketAddress(); , methods: //tab切换 handleClick(tab, event) const _this = this; if (tab.name === 'online') if (!_this.active) if (_this.online > 0) _this.active = _this.friend[0]; _this.activeName = 'online'; _this.receiver = _this.active; _this.getHistoryChat(_this.receiver); else if (_this.userList.length > 0) _this.active = _this.userList[0]; _this.activeName = 'all'; _this.receiver = _this.active; _this.getHistoryChat(_this.receiver); if (tab.name === 'all') if (!_this.active) if (_this.online > 0) _this.active = _this.friend[0]; _this.activeName = 'online'; _this.receiver = _this.active; _this.getHistoryChat(_this.receiver); else if (_this.userList.length > 0) _this.active = _this.userList[0]; _this.activeName = 'all'; _this.receiver = _this.active; _this.getHistoryChat(_this.receiver); , //切换用户 switchUser(data) if (this.active === data) return; this.active = data; this.receiver = data; //获取历史聊天记录 this.getHistoryChat(this.receiver); this.msgNotify = this.msgNotify.filter(item => item != this.active); , //获取历史聊天记录 getHistoryChat(data) this.$http.get('/test/mq/chatMsg/getHistoryChat/' + data).then(res => if (res.data.code === 1) this.chatMsgList = res.data.data; this.flushScroll(); else this.$message.warning(res.data.msg); ).catch(error => this.$message.error(error); ); , //获取websocket地址 getWebSocketAddress() this.$http.get('/test/mq/chatMsg/getWebSocketAddress/' + this.username).then(res => if (res.data.code === 1) if ('WebSocket' in window) this.websocket = new WebSocket(res.data.data); this.initWebSocket(); if (this.username != 'super_admin') this.getHistoryChat(this.username); else this.$message.warning('当前浏览器不支持websocket创建！'); else this.$message.warning(res.data.msg); ).catch(error => this.$message.error(error); ); , //初始化websocket initWebSocket() const _this = this; _this.websocket.onerror = function (event) _this.$message.error('服务端连接错误！'); _this.websocket.onopen = function (event) _this.$message.success("连接成功！"); _this.websocket.onmessage = function (event) let res = JSON.parse(event.data); if (res.key === 1) _this.userList = res.userList; _this.friend = res.onlineList; _this.online = _this.friend.length; if (!_this.active) if (_this.online > 0) _this.active = _this.friend[0]; _this.activeName = 'online'; _this.receiver = _this.active; _this.getHistoryChat(_this.receiver); else if (_this.userList.length > 0) _this.active = _this.userList[0]; _this.activeName = 'all'; _this.receiver = _this.active; _this.getHistoryChat(_this.receiver); if (res.key === 2) _this.userList = res.userList; _this.friend = res.onlineList; _this.online = _this.friend.length; if (!_this.active) if (_this.online > 0) _this.active = _this.friend[0]; _this.activeName = 'online'; _this.receiver = _this.active; _this.getHistoryChat(_this.receiver); else if (_this.userList.length > 0) _this.active = _this.userList[0]; _this.activeName = 'all'; _this.receiver = _this.active; _this.getHistoryChat(_this.receiver); if (res.key === 3) if (_this.username === res.data.sender) _this.chatMsgList.push(res.data); _this.flushScroll(); else if (res.data.sender === 'super_admin') _this.chatMsgList.push(res.data); _this.flushScroll(); else if (res.data.sender === _this.active) _this.chatMsgList.push(res.data); _this.flushScroll(); else //发送其它用户处理 _this.msgNotify.push(res.data.sender); _this.websocket.onclose = function (event) _this.$message.warning('服务端已关闭！'); , //发送消息 sendMsg() if (this.msg.trim().length === 0) this.$message.warning('不能发送空消息！'); return; let chatMsg = ; chatMsg.msg = this.msg; chatMsg.sender = this.username; chatMsg.createTime = new Date(); chatMsg.receiver = this.receiver; this.websocket.send(JSON.stringify(chatMsg)); this.msg = ''; this.flushScroll(); , //刷新滚动条 flushScroll() let content = this.$refs.content; setTimeout(() => content.scrollTop = content.scrollHeight; , 100); , </script> <style scoped lang="scss"> .container padding-top: 50px; .box-card margin: auto; width: 800px; height: 800px; max-height: 900px; ::v-deep .el-card__header background: #867ba9 !important; border-bottom: none; padding: 0; ::v-deep .el-card__body padding: 0px !important; position: relative; .content height: 600px; background: #ddd; overflow-y: auto; .el-divider--horizontal margin: 0; .active color: #0080ff; .mark background: #deb068; .item margin-top: 10px; margin-right: 10px; .operate padding: 5px 15px; </style>
- 启动前后端项目，分别使用客服账号和客户账号登录聊天室
- 聊天消息
结语
至此，关于实现微服务的websocket聊天室到这里就结束了，下期见。。。
(二十一)atp应用测试平台——vue实战之大红灯笼高高挂(代码片段)
前言2022年的最后一天班，好的开始，好的结束。把大红灯笼高高挂起来，欢度元旦的到来，兔年的到来。明年再战。明年再见。。。 正文正菜奉上lantern.vue灯笼源码<template><divclass="app"><divclass... 查看详情
(二十四)atp应用测试平台——springboot集成fastdfs上传与下载功能(代码片段)
前言本节内容我们主要介绍一下如何在springboot项目中集成fastdfs组件，实现文件的上传与下载。关于fastdfs服务中间键的安装过程，本节内容不做介绍。fastdfs是一个轻量级的分布式文件系统，也是我们文件存储中常常... 查看详情
(十四)atp应用测试平台——使用docker-compose一键式安装atp应用测试平台的依赖服务(代码片段)
前言关于ATP应用服务测试平台的相关内容已经更新不少，下载项目的小伙伴第一时间一定是想着怎么把这个平台项目跑起来，看下小编花里胡哨的效果是否能正常show。不过由于依赖的增多，项目的服务也随之多了起... 查看详情
(二十二)atp应用测试平台——swagger2集成swagger-bootstrap-ui实现api文档访问(代码片段)
前言原生的swagger文档虽然也可以作为API文档输出，但是不够简洁优美，本节内容我们介绍一下swagger-bootstrap-ui组件，实现对swaggerAPI文档的二次封装，使用API文档更加简洁明了，无论是入参还是结果响应，... 查看详情
(十八)atp应用测试平台——关于springboot应用监控的那些事(代码片段)
...快告诉我你的绝招。本节内容我们主要介绍一下springboot应用的常见应用参数监控指标，从而更好的关注springboot应用的运行状况并实现应用的监控。除此之外，我们通过集成一 查看详情
(十八)atp应用测试平台——关于springboot应用监控的那些事(代码片段)
...快告诉我你的绝招。本节内容我们主要介绍一下springboot应用的常见应用参数监控指标，从而更好的关注springboot应用的运行状况并实现应用的监控。除此之外，我们通过集成一 查看详情
atp应用测试平台——关于axios的配置使用(代码片段)
前言该篇是应某个粉丝的要求讲解一下关于axios请求组件的使用，其实axios组件类似我们以前使用过的jQuery中的ajax组件，都是用于进行http网络请求的组件。axios组件在前后端分离项目中使用的更加广泛，也更易集成... 查看详情
(十三)atp应用测试平台——springboot集成kafka案例实战(代码片段)
前言kafka消息中间键也是我们经常要用到的功能，尤其是在大数据、高并发的项目中，如日志收集、业务数据分发等等。其最核心的俩大功能作用是:①削峰填谷②异步解耦。本节我们主要介绍一下如何在springboot项目中集... 查看详情
atp应用测试平台——关于网页表格的打印及pdf下载的实战案例(代码片段)
前言在网站应用中，我们可能会有这样一个需求，将网页的部分内容，例如表格，网页片段下载打印或者导出PDF，本小节内容正是关于这样一个内容的实战，基于vue2环境开发，希望能够帮助到你。源码... 查看详情
atp应用测试平台——关于vue-router前端路由的配置使用案例(代码片段)
前言VueRouter是Vue.js(opensnewwindow)官方的路由管理器。它和Vue.js的核心功能深度集成，功能丰富，是我们构建vue前端项目中必不可少的的组件之一，本节我们简单介绍一下在vue项目中如何使用vue-router组件实现页面的路由... 查看详情
atp应用测试平台——关于vue中vue-quill-editormavon-editortinymce等多种富文本编辑器的集成使用(代码片段)
...or以及tinymce等的集成及使用。源代码依然托管在我们的ATP应用测试平台中，源码地址：https://gitee.com/northcangap/atp，仅供参考使用。富文本编辑器实现效果如下： 查看详情
atp应用测试平台——使用bat批处理实现springboot项目的启动与关闭(代码片段)
...停，并且自行配置java环境，这对于windows环境的单应用而言，批处理文件更加简洁，同时也可以实现开机 查看详情
atp应用测试平台——使用easyexcel实现excel导入导出多sheet填充模板下载等功能案例实战(代码片段)
前言Java开发中实现Excel的导入、导出、填充、多sheet页操作等常用功能也是我们经常要面对的开发需求，本文以easyexcel为例，将excel中的常用功能整理成一个个小案例，参考使用。案例源码地址：https://gitee.com/northc... 查看详情
(十七)atp应用测试平台——自定义实现一个springboot2的线程池启动器starter
前言启动器是springboot的一大特点，我们可以根据项目自身需求按需装配我们的组件。例如我们需要操作redis，项目中可以添加一个redis的启动器spring-boot-starter-data-redis，这样redis的一些客户端操作功能我们就集成好了&#... 查看详情
(十七)atp应用测试平台——自定义实现一个springboot2的线程池启动器starter(代码片段)
前言启动器是springboot的一大特点，我们可以根据项目自身需求按需装配我们的组件。例如我们需要操作redis，项目中可以添加一个redis的启动器spring-boot-starter-data-redis，这样redis的一些客户端操作功能我们就集成好了&#... 查看详情
(十九)atp应用测试平台——springboot集成rocketmq案例实战(代码片段)
前言本节内容是关于RocketMQ消息中间键的实战内容，主要介绍在springboot项目中如何集成使用RocketMQ消息中间键，包括消息的发送、消息的接收以及RocketMQ的一些配置说明，以及效果说明。话不多说，开始实战内容。... 查看详情
(十七)atp应用测试平台——redis实现api接口访问限流(固定窗口限流算法)(代码片段)
前言开始正文之前，大多数情况下应该有这样一段场景。面试官：说说平常在项目中，你是如何使用redis的？我：我们就很简单啦，比如前后端分离token的存储、短信验证码的存储，权限列表的存储... 查看详情
atp应用测试平台——使用vue-video-player视频播放组件实现网页视频流的播放案例实战(代码片段)
前言在网页中播放视频也是我们经常要使用到的功能，例如设备监控的视频流实时播放，MP4、m3u8等视频资源播放等等，在vue项目中，我们可以使用目前封装好的开源组件vue-video-player实现上述的要求。本节我们就... 查看详情