百度360必应搜狗淘宝本站头条
当前位置:网站首页 > 技术分类 > 正文

Springboot 整合 Websocket 轻松实现IM及时通讯

ztj100 2025-01-19 02:00 25 浏览 0 评论

一、方案实践

集成分为三步:添加依赖、增加配置类和消息核心类、前端集成。

1.1、添加依赖

Bash
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
    <version>2.1.13.RELEASE</version>
</dependency>

1.2、增加WebSocket配置类

Bash
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;

/**
 * WebSocket配置
 */
@Configuration
public class WebSocketConfig {
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
}

1.3、增加消息核心类WebSocketServer

@ServerEndpoint("/websocket/{userId}")
@Component
@Slf4j
public class WebSocketServer {
    // 消息存储
    private static MessageStore messageStore;
    // 消息发送
    private static MessageSender messageSender;

    public static void setMessageStore(MessageStore messageStore) {
        WebSocketServer.messageStore = messageStore;
    }

    public static void setMessageSender(MessageSender messageSender) {
        WebSocketServer.messageSender = messageSender;
    }

    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("userId") String userId) {
        messageStore.saveSession(session);
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose(Session session, @PathParam("userId") String userId) {
        messageStore.deleteSession(session);
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @ Param message 客户端发送过来的消息
     */
    @OnMessage
    public void onMessage(String message, Session session) throws Exception {
        log.warn("=========== 收到来自窗口" + session.getId() + "的信息:" + message);
        handleTextMessage(session, new TextMessage(message));
    }

    /**
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, @PathParam("userId") String userId, Throwable error) {
        log.error("=========== 发生错误");
        error.printStackTrace();
//        msgStore.deleteSession(session);
    }

    public void handleTextMessage(Session session, TextMessage message) throws Exception {
        log.warn("=========== Received message: {}", message.getPayload());
    }
}

1.4、前端页面加入socket

<!DOCTYPE html>
<html xmlns="http://www.w3.org/1999/html">
  <head>
    <title>WebSocket Example</title>
  </head>
  <body>
    登录用户ID:<input type="text" id="sendUserId" /></br>
    接受用户ID:<input type="text" id="receivedUserId" /></br>
    发送消息内容:<input type="text" id="messageInput" /></br>
    接受消息内容:<input type="text" id="messageReceive" /></br>
    <button onclick="sendMessage()">Send</button>

    <script>
      var socket = new WebSocket("ws://localhost:8080/websocket/aaa");
      var roomId = "123456";
      // 随机产出六位数字
      var sendUserId = Math.floor(Math.random() * 1000000);

      document.getElementById("sendUserId").value = sendUserId;
      var messageReceive = document.getElementById("messageReceive");


      socket.onopen = function (event) {
        console.log("WebSocket is open now.");
        let loginInfo = {
          msgType: 2, //登录消息
          sendUserId: sendUserId,
          bizType: 1, //业务类型
          bizOptModule: 1, //业务模块
          roomId: roomId,
          msgBody: {},
        };
        socket.send(JSON.stringify(loginInfo));
      };

      socket.onmessage = function (event) {
        var message = event.data;
        console.log("Received message: " + message);
        messageReceive.value = message;
      };

      socket.onclose = function (event) {
        console.log("WebSocket is closed now.");
      };

      function sendMessage() {
        var message = document.getElementById("messageInput").value;
        var receivedUserId = document.getElementById("receivedUserId").value;
        let operateInfo = {
          msgType: 100, //业务消息
          sendUserId: sendUserId,
          bizType: 1, //业务类型
          bizOptModule: 1, //业务模块
          roomId: roomId,
          receivedUserId: receivedUserId,
          msgBody: {
            operateType: 1, //操作类型:禁言
            operateContent: "1",
          },
        };
        socket.send(JSON.stringify(operateInfo));
      }

      setInterval(() => {
        socket.send("ping");
      }, 30000);
    </script>
  </body>
</html>


二、小型及时通讯包含的模块

以上只是集成了Websocket框架,实现了基本的全双工通信,服务器和客户端都可以同时发送和接收数据。要想实现一些小型完整的即时通讯,还需要具备以下几个核心模块。架构图如下:

2.1、架构图

2.2、消息对象模型

组织消息内容,比如消息类型、发送者用户ID、接受者用户ID、具体的消息体等。比如:

public class SocketMsg<T> {

    /**
     * 消息类型:1心跳  2登录 3业务操作
     */
    private Integer msgType;

    /**
     * 发送者用户ID
     */
    private String sendUserId;
    /**
     * 接受者用户ID
     */
    private String receivedUserId;

    /**
     * 业务类型
     */
    private Integer bizType;

    /**
     * 业务操作模块
     */
    private Integer bizOptModule;

    /**
     * 消息内容
     */
    private T msgBody;
}

2.3、消息存储模块

负责存储消息内容、用户ID和sessionID的关系,防止数据丢失或者服务器重启等。

2.4、消息发送模块

功能开发完毕,一般部署到分布式集群环境,所以通讯时session会分布在多台服务器。比如用户A的session在机器1,用户B的session在机器2,此时A发送给B,就无法单独通过机器1完成。

因为机器1拿不到机器2里的session,所以消息发不过去。此时只能借助别的中间件来实现,比如借助消息中间件kafka实现。

机器1将消息发送给kafka,然后机器1和机器2都监听kafka,然后查看用户对应的session是否在本机,如果在本机则发送出去。

2.5、消息推送模块

模块3提到的消息发送流程中,消息发送给 消息中间件,然后服务器消费到消费,再通过本机的session推送给客户端。

三、遇到的几个问题

3.1、连接自动断开

WebSocket连接之后,发现一个问题:每隔一段时间如果不传送数据的话,与前端的连接就会自动断开。可以采用心跳消息的方式来解决这个问题。比如客服端每隔30秒自动发送ping消息给服务端,服务端返回pong。

3.2、Session无法被序列化

分布式场景会存在这样的问题:当一次请求负载到第一台服务器时,session在第一台服务器线程上,第二次请求,负载到第二台服务器上,此时通过userId查找当前用户的session时,是查找不到的。

本来想着把session存入到redis中,就可以从redis获取用户的session,希望用这种方式来解决分布式场景下消息发送的问题。但是会出现如下错误:

The remote endpoint was in state [STREAM_WRITING] which is an invalid state for called method

翻看了session源码,发现session无法被序列化。所以这个方案只能放弃。解决方案请看下面的问题5或者上面的架构图

3.3、对象无法自动注入

使用了@ServerEndpoint注解的类中使用@Resource@Autowired注入对象都会失败,并且报空指针异常。

原因是WebSocket服务是线程安全的,那么当我们去发起一个ws连接时,就会创建一个端点对象。WebSocket服务是多对象的,不是单例的。而我们的SpringBean默认就是单例的,在非单例类中注入一个单例的Bean是冲突的。

或者说:

Spring管理采用单例模式(singleton),而 WebSocket 是多对象的,即每个客户端对应后台的一个 WebSocket 对象,也可以理解成 new 了一个 WebSocket,这样当然是不能获得自动注入的对象了,因为这两者刚好冲突。

@Autowired 注解注入对象操作是在启动时执行的,而不是在使用时,而 WebSocket 是只有连接使用时才实例化对象,且有多个连接就有多个对象。所以我们可以得出结论,这个 Service 根本就没有注入到 WebSocket 当中。

如何解决呢?

使用静态对象,并且对外暴露set方法,这样在对象初始化的时候,将其注入到WebSocketServer中。比如说这样:

@ServerEndpoint("/websocket/{userId}")
@Component
@Slf4j
public class WebSocketServer {
  private static MessageStore messageStore;
  private static MessageSender messageSender;

  public static void setMessageStore(MessageStore messageStore) {
      WebSocketServer.messageStore = messageStore;
  }

  public static void setMessageSender(MessageSender messageSender) {
      WebSocketServer.messageSender = messageSender;
  }
}


@Slf4j
@Service
public class MessageStore {
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @PostConstruct
    public void init() {
        WebSocketServer.setMessageStore(this);
    }
}

3.4、分布式场景消息如何发给客户端

问题2中提到了分布式场景下存在的session不在本机的问题,这种场景可以通过发送消息中间件的方式解决。具体这样解决:

每次连接时,都将userId和对应的session存入到本机,发送消息时,直接发送给MQ-Broker,然后每台应用负载都去消费这个消息,拿到消息之后,判断在本机能根据userId是否能找到session,找到session则推送到客户端。

3.5、部署时Nginx配置问题

代码开发完毕之后,本机跑通后,然后部署到服务器之后,还差很重要的一步:配置nginx代理。

3.5.1、给后端应用部署独立域名

要给后端应用部署独立域名,nginx代理直接转发到应用的独立域名,不要走微服务的gateway网关转发过去。

3.5.2、多层nginx转发问题

当只有一层nginx的时候,配置比较简单,如下:

location ~* ^/api/websocket/* {
      proxy_pass http://mangodwsstest.mangod.top;
      
      proxy_read_timeout 300s;
      proxy_send_timeout 300s;
      proxy_set_header Host mangodwsstest.mangod.top;
      proxy_http_version 1.1;
      proxy_set_header Upgrade $http_upgrade;
      proxy_set_header Connection "Upgrade";
      proxy_set_header X-Real-IP $remote_addr;
 }

但是,当有两层nginx转发的时候,问题就出现了。

在最外层的nginx需要使用如下配置,不能照抄后面一层的配置。proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_forproxy_set_header X-Forwarded-Proto $scheme这两个配置不能少,用来将协议和真实IP传递给后面一层的nginx。

location ~* ^/api/websocket/* {
      proxy_pass http://mangodwsstest.mangod.top;

      proxy_read_timeout 300s;
      proxy_send_timeout 300s;
      proxy_set_header  Host $http_host;
      proxy_set_header  X-Real-IP  $remote_addr;
      proxy_set_header  X-Forwarded-For $proxy_add_x_forwarded_for;
      proxy_set_header  X-Forwarded-Proto $scheme;
      proxy_http_version 1.1;
      proxy_set_header Upgrade $http_upgrade;
      proxy_set_header Connection $connection_upgrade;
}

四、完整代码和示例

4.1、页面效果如下

开启两个web页面,用户1输入用户2的用户ID,输入发送消息内容,点击发送。在用户2的页面的接受消息内容可以看到发送的消息。

4.2、代码结构

4.3、代码地址

https://github.com/yclxiao/spring-websocket.git

五、总结

本文聊了Springboot如何集成Websocket、IM及时通讯需要哪些模块、开发和部署过程中遇到的问题、以及实现小型IM及时通讯的代码。希望对你有帮助!

相关推荐

Python 操作excel的坑__真实的行和列

大佬给的建议__如何快速处理excelopenpyxl库操作excel的时候,单个表的数据量大一些处理速度还能接受,如果涉及多个表甚至多个excel文件的时候速度会很慢,还是建议用pandas来处理,...

Python os.path模块使用指南:轻松处理文件路径

前言在Python编程中,文件和目录的操作是非常重要的一部分。为了方便用户进行文件和目录的操作,Python标准库提供了os模块。其中,os.path子模块提供了一些处理文件路径的函数和方法。本文主要...

Python常用内置模块介绍——文件与系统操作详解

Python提供了多个强大的内置模块用于文件和系统操作,下面我将详细介绍最常用的几个模块及其核心功能。1.os模块-操作系统交互...

Python Flask 建站框架实操教程(flask框架网页)

下面我将带您从零开始构建一个完整的Flask网站,包含用户认证、数据库操作和前端模板等核心功能。##第一部分:基础项目搭建###1.创建项目环境```bash...

为你的python程序上锁:软件序列号生成器

序列号很多同学可能开发了非常多的程序了,并且进行了...

PO设计模式全攻略,在 UI 自动化中的实践总结(以企业微信为例)

一、什么是PO设计模式?PO(PageObject)设计模式将某个页面的所有元素对象定位和对元素对象的操作封装成一个Page类,即一个py文件,并以页面为单位来写测试用例,实现页面对象和测试用例的...

这种小工具居然也能在某鱼卖钱?我用Python一天能写...

前两天在某鱼闲逛,本来想找个二手机械键盘,结果刷着刷着突然看到有人在卖——Word批量转PDF小工具...

python打包成exe,程序有图标,但是任务栏和窗口都没有显示图标

代码中指定图标信息#设置应用ID,确保任务栏图标正确显示ifsys.platform=="win32":importctypesapp_id=...

使用Python构建电影推荐系统(用python做推荐系统)

在日常数据挖掘工作中,除了会涉及到使用Python处理分类或预测任务,有时候还会涉及推荐系统相关任务。...

python爬取并分析淘宝商品信息(python爬取淘宝商品数据)

python爬取并分析淘宝商品信息背景介绍一、模拟登陆二、爬取商品信息1.定义相关参数2.分析并定义正则3.数据爬取三、简单数据分析1.导入库2.中文显示3.读取数据4.分析价格分布5.分析销售...

OpenCV入门学习基础教程(从小白变大神)

Opencv是用于快速处理图像处理、计算机视觉问题的工具,支持多种语言进行开发如c++、python、java等,下面这篇文章主要给大家介绍了关于openCV入门学习基础教程的相关资料,需要的朋友可以...

python图像处理-一行代码实现灰度图抠图

抠图是ps的最基本技能,利用python可以实现用一行代码实现灰度图抠图。基础算法是...

从头开始学python:如何用Matplotlib绘图表

Matplotlib是一个用于绘制图表的库。如果你有用过python处理数据,那Matplotlib可以更直观的帮你把数据展示出来。直接上代码看例子:importmatplotlib.pyplot...

Python爬取爱奇艺腾讯视频 250,000 条数据分析为什么李诞不值得了

在《Python爬取爱奇艺52432条数据分析谁才是《奇葩说》的焦点人物?》这篇文章中,我们从爱奇艺爬取了5万多条评论数据,并对一些关键数据进行了分析,由此总结出了一些明面上看不到的数据,并...

Python Matplotlib 库使用基本指南

简介Matplotlib是一个广泛使用的Python数据可视化库,它可以创建各种类型的图表、图形和可视化效果。无论是简单的折线图还是复杂的热力图,Matplotlib提供了丰富的功能来满足我们...

取消回复欢迎 发表评论: