• 基于 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>

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6

    2、属性配置

    然后我们对其进行配置

    首先是 yml 的配置

    # netty-socketio 配置
socketio:
  host: 0.0.0.0
  port: 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 # 设置最大每帧处理数据的长度，防止他人利用大数据来攻击服务器

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11

    然后是 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);//指定传输协议为WebSocket
        return new SocketIOServer(config);
    }

    /**
     * 注册socket-io 注解
     * 开启SocketIOServer注解支持，比如 @OnConnect、@OnEvent
     * @param socketServer socketIo 服务
     * @return
     */
    @Bean
    public SpringAnnotationScanner springAnnotationScanner(SocketIOServer socketServer) {
        return new SpringAnnotationScanner(socketServer);
    }

}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88
    • 89
    • 90
    • 91
    • 92
    • 93
    • 94
    • 95
    • 96
    • 97
    • 98
    • 99
    • 100

    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;
    }

    @Override
    public void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent) {

        server.start();
        log.info("--------------------------------------");
        log.info("socket.io启动成功: " + server.getConfiguration().getPort());
        log.info("--------------------------------------");
    }
}



    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34

    然后是关闭时的监听器

    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
    @Override
    public void onApplicationEvent(ContextClosedEvent contextClosedEvent) {

        server.stop();
        log.info("--------------------------------------");
        log.info("socket.io 关闭成功");
        log.info("--------------------------------------");
    }
}



    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35

    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 {

    @Resource
    private SocketIOServer socketIOServer;


    //socket添加@OnDisconnect事件，客户端断开连接时调用，刷新客户端信息
    @OnDisconnect
    public void onDisconnect(SocketIOClient client) {
    }

    //socket添加connect事件，当客户端发起连接时调用
    @OnConnect
    public void onConnect(SocketIOClient client) {
    }

    //接收到新的消息对消息类型所属的会话进行判断
    @OnEvent("sendNewMessage")
    public void sendNewMessage(SocketIOClient client, NewMessageVo newMessageVo) 
    }

}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43

    我的服务，我这里是基于 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 {

    @Resource
    private SocketIOServer socketIOServer;

    @Resource
    private SingleMessageService singleMessageService;

    @Resource
    private OnlineUserService onlineUserService;

    @Resource
    private GroupMessageService groupMessageService;

    /**
     * map：clientId -> uid（用户判断用户上下线）
     * map：uid -> simpleUser（用于查询是否已经登录）
     */

    //socket添加@OnDisconnect事件，客户端断开连接时调用，刷新客户端信息
    @OnDisconnect
    public 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事件，当客户端发起连接时调用
    @OnConnect
    public 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());
    }

}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88
    • 89
    • 90
    • 91
    • 92
    • 93
    • 94
    • 95
    • 96
    • 97
    • 98
    • 99
    • 100
    • 101
    • 102
    • 103
    • 104
    • 105
    • 106
    • 107
    • 108
    • 109
    • 110
    • 111
    • 112
    • 113
    • 114
    • 115
    • 116
    • 117
    • 118
    • 119
    • 120
    • 121
    • 122
    • 123
    • 124
    • 125
    • 126
    • 127
    • 128
    • 129
    • 130
    • 131
    • 132
    • 133
    • 134
    • 135
    • 136
    • 137
    • 138

    5、前端

    下载包

    npm install vue-socket.io@3.0.10

    • 1

    main.js配置

    import VueSocketIO from 'vue-socket.io'

Vue.use(new VueSocketIO({
  debug: true,
  connection: 'http://localhost:9999'
}))


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7

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

      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)
    }
  },

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 发送消息，我们直接采用 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})
    },

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18

    接受消息

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

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
  • 相关阅读:
    安装ThinkPHP5.1并在框架中使用FFmpeg视频处理工具遇到的问题和解决办法
    C:入门级积累（5）
    转转品牌升级，不只换了个LOGO
    Netty 学习（六）：创建 NioEventLoopGroup 的核心源码说明
    java培训技术自定义视图介绍
    搜索技术【二分搜索】 - 简介 & 原理
    PostgreSQL+GeoHash地图点位聚合
    Windows配置JAVA_HOME但是未生效问题
    C/C++总结笔记—— volatile、mutable、explicit 关键字
    第3章业务功能开发（修改市场活动备注）
  • 原文地址：https://blog.csdn.net/qq_45803593/article/details/125525629