基于 socketio 的 room 的使用

基于 socketio 的 room 的使用

这里我们直接讲解基于 springboot + vue 的 socketio 服务代码，基于 socketio 我们可以使得服务端直接向客户端进行消息的推送，而不再是只能客户端请求服务端响应的模式，基于此 socketio 常常用于即时通讯，我们基于聊天的场景，使用其 room 来简单介绍聊天项目的编写。

1、导入jar

首先我们在后端导入jar

        <!-- socket.io服务端 --><dependency><groupId>com.corundumstudio.socketio</groupId><artifactId>netty-socketio</artifactId><version>1.7.7</version></dependency>

2、属性配置

然后我们对其进行配置

首先是 yml 的配置

# netty-socketio 配置
socketio:host: 0.0.0.0port: 9999 # 端口号workCount: 100 # 工作线程数allowCustomRequests: true # 允许服务自定义请求与socket.io协议不同。upgradeTimeout: 10000 # 协议升级超时时间（毫秒），默认10秒。HTTP握手升级为ws协议超时时间pingTimeout: 60000 # Ping消息超时时间（毫秒），默认60秒，这个时间间隔内没有接收到心跳消息就会发送超时事件pingInterval: 25000 # Ping消息间隔（毫秒），默认25秒。客户端向服务器发送一条心跳消息间隔maxHttpContentLength: 1048576 # 设置http交互最大内容长度maxFramePayloadLength: 1048576 # 设置最大每帧处理数据的长度，防止他人利用大数据来攻击服务器

然后是 java 的配置

package com.manster.server.config;import com.corundumstudio.socketio.SocketIOServer;
import com.corundumstudio.socketio.Transport;
import com.corundumstudio.socketio.annotation.SpringAnnotationScanner;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** 配置 socketio* @Author manster* @Date 2022/6/13**/
@Slf4j
@Configuration
public class SocketIoConfig {/*** socketio 地址*/@Value("${socketio.host}")private String host;/*** socketio 端口*/@Value("${socketio.port}")private Integer port;/*** 工作线程*/@Value("${socketio.workCount}")private int workCount;/*** 允许服务自定义请求与socket.io协议不同。*/@Value("${socketio.allowCustomRequests}")private boolean allowCustomRequests;/*** 协议升级超时时间（毫秒），默认10秒。HTTP握手升级为ws协议超时时间*/@Value("${socketio.upgradeTimeout}")private int upgradeTimeout;/**** Ping消息超时时间（毫秒），默认60秒，这个时间间隔内没有接收到心跳消息就会发送超时事件*/@Value("${socketio.pingTimeout}")private int pingTimeout;/*** Ping消息间隔（毫秒），默认25秒。客户端向服务器发送一条心跳消息间隔*/@Value("${socketio.pingInterval}")private int pingInterval;@Value("${socketio.maxFramePayloadLength}")private int maxFramePayloadLength;@Value("${socketio.maxHttpContentLength}")private int maxHttpContentLength;@Bean("socketIOServer")public SocketIOServer socketIOServer() {com.corundumstudio.socketio.Configuration config = new com.corundumstudio.socketio.Configuration();config.setHostname(host);config.setPort(port);com.corundumstudio.socketio.SocketConfig socketConfig = new com.corundumstudio.socketio.SocketConfig();socketConfig.setReuseAddress(true);config.setSocketConfig(socketConfig);config.setWorkerThreads(workCount);config.setAllowCustomRequests(allowCustomRequests);config.setUpgradeTimeout(upgradeTimeout);config.setPingTimeout(pingTimeout);config.setPingInterval(pingInterval);config.setMaxHttpContentLength(maxHttpContentLength);config.setMaxFramePayloadLength(maxFramePayloadLength);config.setTransports(Transport.WEBSOCKET);//指定传输协议为WebSocketreturn new SocketIOServer(config);}/*** 注册socket-io 注解* 开启SocketIOServer注解支持，比如 @OnConnect、@OnEvent* @param socketServer socketIo 服务* @return*/@Beanpublic SpringAnnotationScanner springAnnotationScanner(SocketIOServer socketServer) {return new SpringAnnotationScanner(socketServer);}}

3、配置监听器

首先是初始化的监听器

package com.manster.server.listen;import com.corundumstudio.socketio.SocketIOServer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;/*** 项目启动成功后，启动socket-io 服务* @Author manster* @Date 2022/6/13**/
@Slf4j
@Component
public class SocketIoInitListener implements ApplicationListener<ApplicationReadyEvent> {private final SocketIOServer server;public SocketIoInitListener(SocketIOServer server) {this.server = server;}@Overridepublic void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent) {server.start();log.info("--------------------------------------");log.info("socket.io启动成功: " + server.getConfiguration().getPort());log.info("--------------------------------------");}
}

然后是关闭时的监听器

package com.manster.server.listen;import com.corundumstudio.socketio.SocketIOServer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextClosedEvent;
import org.springframework.stereotype.Component;/*** 项目关闭后，关闭socket-io 服务* @Author manster* @Date 2022/6/13**/
@Slf4j
@Component
public class SocketIoStopListener implements ApplicationListener<ContextClosedEvent> {private final SocketIOServer server;public SocketIoStopListener(SocketIOServer server) {this.server = server;}// 监听kill pid     无法监听 kill -9 pid@Overridepublic void onApplicationEvent(ContextClosedEvent contextClosedEvent) {server.stop();log.info("--------------------------------------");log.info("socket.io 关闭成功");log.info("--------------------------------------");}
}

4、socketio 服务

配置 socketio 的消息服务

核心代码

package com.manster.server.handler;import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.corundumstudio.socketio.SocketIOClient;
import com.corundumstudio.socketio.SocketIOServer;
import com.corundumstudio.socketio.annotation.OnConnect;
import com.corundumstudio.socketio.annotation.OnDisconnect;
import com.corundumstudio.socketio.annotation.OnEvent;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;import javax.annotation.Resource;
import java.util.*;/*** @Author manster* @Date 2022/6/13**/
@Slf4j
@Component
public class SocketIOHandler {@Resourceprivate SocketIOServer socketIOServer;//socket添加@OnDisconnect事件，客户端断开连接时调用，刷新客户端信息@OnDisconnectpublic void onDisconnect(SocketIOClient client) {}//socket添加connect事件，当客户端发起连接时调用@OnConnectpublic void onConnect(SocketIOClient client) {}//接收到新的消息对消息类型所属的会话进行判断@OnEvent("sendNewMessage")public void sendNewMessage(SocketIOClient client, NewMessageVo newMessageVo) }}

我的服务，我这里是基于 socketio 中 room 来进行会话的，私聊就是两人一个房间（房间号为两人1-2，两人id相连接），群聊就以群聊id为房间号，聊天前需要先是客户端加入房间

package com.manster.server.handler;import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.corundumstudio.socketio.SocketIOClient;
import com.corundumstudio.socketio.SocketIOServer;
import com.corundumstudio.socketio.annotation.OnConnect;
import com.corundumstudio.socketio.annotation.OnDisconnect;
import com.corundumstudio.socketio.annotation.OnEvent;
import com.manster.server.common.ConstValueEnum;
import com.manster.server.entity.GroupMessage;
import com.manster.server.entity.SingleMessage;
import com.manster.server.entity.User;
import com.manster.server.service.GroupMessageService;
import com.manster.server.service.OnlineUserService;
import com.manster.server.service.SingleMessageService;
import com.manster.server.vo.CurrentConversationVo;
import com.manster.server.vo.NewMessageVo;
import com.manster.server.vo.SimpleUser;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Component;import javax.annotation.Resource;
import java.util.*;/*** @Author manster* @Date 2022/6/13**/
@Slf4j
@Component
public class SocketIOHandler {@Resourceprivate SocketIOServer socketIOServer;@Resourceprivate SingleMessageService singleMessageService;@Resourceprivate OnlineUserService onlineUserService;@Resourceprivate GroupMessageService groupMessageService;/*** map：clientId -> uid（用户判断用户上下线）* map：uid -> simpleUser（用于查询是否已经登录）*///socket添加@OnDisconnect事件，客户端断开连接时调用，刷新客户端信息@OnDisconnectpublic void onDisconnect(SocketIOClient client) {Map<String, List<String>> urlParams = client.getHandshakeData().getUrlParams();log.info("--------------------客户端已断开连接--------------------{}", urlParams);//清除用户登录信息cleanLoginInfo(client.getSessionId().toString());socketIOServer.getBroadcastOperations().sendEvent("onlineUser", onlineUserService.getOnlineUidSet());}//socket添加connect事件，当客户端发起连接时调用@OnConnectpublic void onConnect(SocketIOClient client) {Map<String, List<String>> urlParams = client.getHandshakeData().getUrlParams();log.info("链接开启，urlParams：{}", urlParams);}//用户上线了@OnEvent("goOnline")public void goOnline(SocketIOClient client, User user) {log.info("goOnline ---> user：{}", user);String clientId = client.getSessionId().toString();SimpleUser simpleUser = new SimpleUser();BeanUtils.copyProperties(user, simpleUser);onlineUserService.addClientIdToSimpleUser(clientId, simpleUser);//广播所有在线用户socketIOServer.getBroadcastOperations().sendEvent("onlineUser", onlineUserService.getOnlineUidSet());}//用户下线了@OnEvent("leave")public void leave(SocketIOClient client) {log.info("leave ---> client：{}", client);//清除用户登录信息cleanLoginInfo(client.getSessionId().toString());//广播所有在线用户socketIOServer.getBroadcastOperations().sendEvent("onlineUser", onlineUserService.getOnlineUidSet());}// 清除用户登录信息private void cleanLoginInfo(String clientId) {SimpleUser simpleUser = onlineUserService.getSimpleUserByClientId(clientId);if (simpleUser != null) {onlineUserService.removeClientAndUidInSet(clientId, simpleUser.getId());}}//接收到新的消息对消息类型所属的会话进行判断@OnEvent("sendNewMessage")public void sendNewMessage(SocketIOClient client, NewMessageVo newMessageVo) {log.info("sendNewMessage ---> newMessageVo：{}", newMessageVo);if (Objects.equals(newMessageVo.getConversationType(), ConstValueEnum.SINGLE)) {SingleMessage singleMessage = new SingleMessage();BeanUtils.copyProperties(newMessageVo, singleMessage);singleMessage.setSenderId(newMessageVo.getSenderId());//插入数据库singleMessageService.save(singleMessage);} else if (Objects.equals(newMessageVo.getConversationType(), ConstValueEnum.GROUP)) {GroupMessage groupMessage = new GroupMessage();BeanUtils.copyProperties(newMessageVo, groupMessage);groupMessage.setSenderId(newMessageVo.getSenderId());//插入数据库groupMessageService.save(groupMessage);}//通知该房间收到消息接受到消息Collection<SocketIOClient> clients = socketIOServer.getRoomOperations(newMessageVo.getRoomId()).getClients(); //实际上同一房间只有2个客户端for (SocketIOClient item : clients) {if (item != client) {item.sendEvent("receiveMessage", newMessageVo);if(singleMessageService.count(new QueryWrapper<SingleMessage>().eq("room_id", newMessageVo.getRoomId())) == 1){item.sendEvent("newConversation", newMessageVo);log.info("加入会话");}}}}@OnEvent("join")public void join(SocketIOClient client, CurrentConversationVo conversationVo) {log.info("加入房间号码：{} ---> conversationVo：{}", conversationVo.getRoomId(), conversationVo);//当前登录的客户端加入到指定房间client.joinRoom(conversationVo.getRoomId());}}

5、前端

下载包

npm install vue-socket.io@3.0.10

main.js配置

import VueSocketIO from 'vue-socket.io'Vue.use(new VueSocketIO({debug: true,connection: 'http://localhost:9999'
}))

在登录后我们进行客户端的连接

  sockets: {// 客户端connect事件，服务端可针对connect进行信息传输connect: function () {this.$message.info('连接成功')console.log('socket connected：', this.$socket.id)},onlineUser (data) {// console.log('当前在线用户列表：', data)this.$store.dispatch('user/SET_ONLINE_USER', data)}},
  • 发送消息，我们直接采用 this.$socket.emit('sendNewMessage', newMessage)，sendNewMessage 为后端接口中 @OnEvent("sendNewMessage") 相对应
  • 接收消息，我们在 vue 的 sockets 钩子中，编写相对应后端发送的 item.sendEvent("newConversation", newMessageVo); 方法，并对应进行数据的接受

然后我们进行消息的发送

    send (e) {e.preventDefault()if (!this.messageText) {return}const common = this.generatorMessageCommon()const newMessage = {...common,message: this.messageText,messageType: 0}this.messages = [...this.messages, newMessage]this.$socket.emit('sendNewMessage', newMessage)this.messageText = ''this.showEmojiCom = false// 发完消息更新自己会话列表的最新消息this.$store.dispatch('conversation/SET_UNREAD_NUM', {type: 'clear', data: newMessage})},

接受消息

  sockets: {receiveMessage (news) {// 收到消息if (news.roomId === this.currentConversation.roomId) {// 是自己正所处的房间就添加消息并更新会话列表this.messages = [...this.messages, news]// 进来设置该房间未读消息数为0setTimeout(() => {this.$store.dispatch('conversation/SET_UNREAD_NUM', {type: 'clear', data: news})}, 0)} else {// 不是现在所处的房间就新增未读消息this.$store.dispatch('conversation/SET_UNREAD_NUM', {type: 'add', data: news})}}},

Published by

风君子

独自遨游何稽首 揭天掀地慰生平

发表回复

您的电子邮箱地址不会被公开。 必填项已用 * 标注