SpringBoot+Netty+Vue+Websocket实现在线推送/聊天系统

Posted 2022-11-24 Huterox

文章目录

• 前言
• 后端
• • 项目结构
  • 初始化器
  • 服务类
• 前端
• • 封装websocket
  • 使用
• 效果

前言

ok，那么今天的话也是带来这个非常常用的一个技术，那就是咱们完成nutty的一个应用，今天的话，我会介绍地很详细，这样的话，拿到这个博文的代码就基本上可以按照自己的想法去构建自己的一个在线应用了。比如聊天，在线消息推送之类的。其实一开始我原来的想法做在线消息推送是直接mq走起，但是想了想对mq的依赖太高了。而且总感觉不安全，况且还有实时在线处理的一些要求，所以的话才觉得切换nutty来做。我的构想是这样的：

在我的构想里面的话，基本上除了和客户端建立的连接之外，会暴露出我们的一个服务器地址和接口。
其他的业务服务，都是通过其他的服务进行调用后返回的，客户端和nutty服务器只是建立长连接，负责接收消息，确认消息。具体的业务消息是如何发送的都是通过其他微服务的，好处就是确保安全，例如限制用户的聊天评率（因为可能是恶意脚本）。

不过的话，我们今天的部分是在这里：

就是紫色框起来的地方。这部分是基础，也是毛坯房，后面你们可以根据本文去造自己的房子。

后端

首先是我们的服务后端的搭建，这部分的话其实可以参考我的这篇文章：实用水文篇–SpringBoot整合Netty实现消息推送服务器

那么我们这边只是说说不同的地方，核心的主要的地方。

项目结构

这里的话，可以看到我们的这边的话其实是和先前的一样的，其实没什么变化，区别在里面：

这里面我重写了一下方法，对上次的一些内容进行了修改，因为上次是毛坯中的毛坯嘛。

初始化器

首先是我们的初始化器，那么在这里的话，我增加了这个心跳在线的一个处理。主要是因为，实际上，就是说，避免我们的一个资源的浪费嘛。

public class ServerHandler extends ChannelInitializer<SocketChannel> 

    /**
     * 初始化通道以及配置对应管道的处理器
     * @param channel
     * @throws Exception
     */
    @Override
    protected void initChannel(SocketChannel channel) throws Exception
   
        ChannelPipeline pipeline = channel.pipeline();
        pipeline.addLast(new HttpServerCodec());
        
        pipeline.addLast(new ChunkedWriteHandler());
        pipeline.addLast(new HttpObjectAggregator(1024*64));

        //===========================增加心跳支持==============================

        /**
         * 针对客户端，如果在1分钟时间内没有向服务端发送读写心跳（ALL），则主动断开连接
         * 如果有读空闲和写空闲，则不做任何处理
         */
        pipeline.addLast(new IdleStateHandler(8,10,12));
        //自定义的空闲状态检测的handler
        pipeline.addLast(new HeartBeatHandler());

      
        pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));

        //自定义的handler
        pipeline.addLast(new ServerListenerHandler());


    

对应的心跳检测的实现类在这里：

public class HeartBeatHandler extends ChannelInboundHandlerAdapter 

    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception 
        if(evt instanceof IdleStateEvent)
            IdleStateEvent event = (IdleStateEvent)evt;//强制类型转化
            if(event.state()== IdleState.READER_IDLE)
                System.out.println("进入读空闲......");
            else if(event.state() == IdleState.WRITER_IDLE) 
                System.out.println("进入写空闲......");
            else if(event.state()== IdleState.ALL_IDLE)
                System.out.println("channel 关闭之前：users 的数量为："+ UserConnectPool.getChannelGroup().size());
                Channel channel = ctx.channel();
                //资源释放
                channel.close();
                System.out.println("channel 关闭之后：users 的数量为："+UserConnectPool.getChannelGroup().size());
            
        
    

服务类

之后的话就是我们具体的消息推送，服务之类的了。

public class ServerListenerHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> 

    private static final Logger log = LoggerFactory.getLogger(ServerBoot.class);

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception 
        //获取客户端所传输的消息
        String content = msg.text();
        //1.获取客户端发来的消息
        DataContent dataContent = JsonUtils.jsonToPojo(content, DataContent.class);
        assert dataContent != null;
        System.out.println("----->"+dataContent);
        Integer action = dataContent.getAction();
        Channel channel =  ctx.channel();

        //2.判断消息类型，根据不同的类型来处理不同的业务
        if(Objects.equals(action, MessageActionEnum.CONNECT.type))
            //2.1 当websocket 第一次open的时候，初始化channel，把用的channel 和 userid 关联起来
            String senderId = dataContent.getChatMsg().getSenderId();
            UserConnectPool.getChannelMap().put(senderId,channel);
            //这里是输出一个用户关系
            UserConnectPool.output();
         

         else if(Objects.equals(action, MessageActionEnum.KEEPALIVE.type))
            //2.4 心跳类型的消息
            System.out.println("收到来自channel 为["+channel+"]的心跳包");
        

    

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception 
        //接收到请求
        log.info("有新的客户端链接：[]", ctx.channel().id().asLongText());
        UserConnectPool.getChannelGroup().add(ctx.channel());
    

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception 
        String chanelId = ctx.channel().id().asShortText();
        log.info("客户端被移除：channel id 为："+chanelId);
        UserConnectPool.getChannelGroup().remove(ctx.channel());
    

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception 
        cause.printStackTrace();
        //发生了异常后关闭连接，同时从channelgroup移除
        ctx.channel().close();
        UserConnectPool.getChannelGroup().remove(ctx.channel());
    

可以看到这里只保留了两个玩意，一个是把用户注册到咱们的这个nutty服务器的内存里面。还有一个是心跳包。

那么其他的数据类型什么的，都在整合nutty的那篇博文里面。

那么我们的聊天怎么处理，很简单，在Controller接受到消息，然后在那里面调用Channel完成消息的转发。

具体的案例也在那篇nutty的整合里面。

前端

那么之后的话，是我们的一个前端 。

封装websocket

这边的话对这个websocket做了一个封装，可以在vue、uniapp当中使用。我这边还用到了element-ui主要是来做消息提醒的，你可以选择删掉。

// 导出socket对象
export 
  socket

import  Message  from 'element-ui'
// socket主要对象
var socket = 
  websock: null,
  /**
   * 这个是我们的ws的地址
   * */
  ws_url: "ws://localhost:9000/ws",
  /**
   * 开启标识
   * */
  socket_open: false,
  /**
   * 心跳timer
   * */
  hearbeat_timer: null,
  /**
   * 心跳发送频率
   * */
  hearbeat_interval: 5000,
  /**
   * 是否开启重连
   * */
  is_reonnect: true,
  /**
   * 重新连接的次数
   * */
  reconnect_count: 3,
  /**
   * 当前重新连接的次数，默认为：1
   * */
  reconnect_current: 1,
  /**
   * 重新连接的时间类型
   * */
  reconnect_timer: null,
  /**
   * 重新连接的间隔
   * */
  reconnect_interval: 3000,

  /**
   * 初始化连接
   */
  init: () => 
    if (!("WebSocket" in window)) 
      Message(
        message: '当前浏览器与网站不兼容丫',
        type: 'error',
      );
      console.log('浏览器不支持WebSocket')
      return null
    

    // 已经创建过连接不再重复创建
    if (socket.websock) 
      return socket.websock
    

    socket.websock = new WebSocket(socket.ws_url)
    socket.websock.onmessage = function (e) 
      socket.receive(e)
    

    // 关闭连接
    socket.websock.onclose = function (e) 
      console.log('连接已断开')
      console.log('connection closed (' + e.code + ')')
      clearInterval(socket.hearbeat_interval)
      socket.socket_open = false

      // 需要重新连接
      if (socket.is_reonnect) 
        socket.reconnect_timer = setTimeout(() => 
          // 超过重连次数
          if (socket.reconnect_current > socket.reconnect_count) 
            clearTimeout(socket.reconnect_timer)
            return
          

          // 记录重连次数
          socket.reconnect_current++
          socket.reconnect()
        , socket.reconnect_interval)
      
    

    // 连接成功
    socket.websock.onopen = function () 
      Message(
        message: '连接成功，欢迎来到WhiteHole',
        type: 'success',
      );
      console.log('连接成功')
      socket.socket_open = true
      socket.is_reonnect = true
      // 开启心跳
      socket.heartbeat()
    
    // 连接发生错误
    socket.websock.onerror = function (err) 
      Message(
        message: '服务连接发送错误！',
        type: 'error',
      );
      console.log('WebSocket连接发生错误')
    
  ,
  /**
   * 获取websocket对象
   * */

  getSocket:()=>
    //创建了直接返回，反之重来
    if (socket.websock) 
      return socket.websock
    else 
      socket.init();
    
  ,

  getStatus:()=> 
    if (socket.websock.readyState === 0) 
      return "未连接";
     else if (socket.websock.readyState === 1) 
      return "已连接";
     else if (socket.websock.readyState === 2) 
      return "连接正在关闭";
     else if (socket.websock.readyState === 3) 
      return "连接已关闭";
    
  ,

  /**
   * 发送消息
   * @param * data 发送数据
   * @param * callback 发送后的自定义回调函数
   */
  send: (data, callback = null) => 
    // 开启状态直接发送
    if (socket.websock.readyState === socket.websock.OPEN) 
      socket.websock.send(JSON.stringify(data))

      if (callback) 
        callback()
      

      // 正在开启状态，则等待1s后重新调用
     else if (socket.websock.readyState === socket.websock.CONNECTING) 
      setTimeout(function () 
        socket.send(data, callback)
      , 1000)

      // 未开启，则等待1s后重新调用
     else 
      socket.init()
      setTimeout(function () 
        socket.send(data, callback)
      , 1000)
    
  ,

  /**
   * 接收消息
   * @param * message 接收到的消息
   */
  receive: (message) => 
    var recData = JSON.parse(message.data)
    /**
     *这部分是我们具体的对消息的处理
     * */
    // 自行扩展其他业务处理...
  ,

  /**
   * 心跳
   */
  heartbeat: () => 
    console.log('socket', 'ping')
    if (socket.hearbeat_timer) 
      clearInterval(socket.hearbeat_timer)
    

    socket.hearbeat_timer = setInterval(() => 
      //发送心跳包
      let data = 
        "action": 4,
        "chatMsg": null,
        "extend": null,
      
      socket.send(data)
    , socket.hearbeat_interval)
  ,

  /**
   * 主动关闭连接
   */
  close: () => 
    console.log('主动断开连接')
    clearInterval(socket.hearbeat_interval)
    socket.is_reonnect = false
    socket.websock.close()
  ,

  /**
   * 重新连接
   */
  reconnect: () => 
    console.log('发起重新连接', socket.reconnect_current)

    if (socket.websock && socket.socket_open) 
      socket.websock.close()
    

    socket.init()
  ,

使用

这个使用其实很简单，我们这边的话是Vue所以在开启的时候就用上了，在我们的这个App.vue或者是其他的主页面里面，我这边是home作为主页面（App.vue直接展示了home.vue(这个是你自己编写的)）

效果

刚刚的连接效果看到了，那么就来看到这个，我们后端的一个心跳：

可以看到以前正常。
之后的话，拿着这套毛坯房就可以happy了。