SpringBoot集成WebSocket实现在线聊天

前言

在项目过程中涉及到了在线聊天的业务，刚好有了解到WebSocket可以实现这一功能，因此便对其进行了一定的研究并做下笔记，在本文中主要借鉴了以下资源：

• WebSocket_百度百科
• 李士伟的小程序聊天工程
• Springboot+Websocket中@Autowired注入service为null的解决方法

1、WebSocket引入

WebSocket是HTML5规范中的一个部分，它借鉴了socket这种思想，为web应用程序客户端和服务端之间提供了一种全双工通信机制。同时，它又是一种新的应用层协议，WebSocket协议是为了提供web应用程序和服务端全双工通信而专门制定的一种应用层协议，其基于TCP传输协议，并复用了HTTP的握手通道。通常它表示为：ws://echo.websocket.org/?encoding=text HTTP/1.1，可以看到除了前面的协议名和http不同之外，它的表示地址就是传统的url地址。

WebSocket的优点主要如下：

1. 支持双向通信，实时性更强。
2. 更好的二进制支持。
3. 较少的控制开销。连接创建后，ws客户端、服务端进行数据交换时，协议控制的数据包头部较小。在不包含头部的情况下，服务端到客户端的包头只有2~10字节（取决于数据包长度），客户端到服务端的的话，需要加上额外的4字节的掩码。而HTTP协议每次通信都需要携带完整的头部。
4. 支持扩展。ws协议定义了扩展，用户可以扩展协议，或者实现自定义的子协议。（比如支持自定义压缩算法等）
5. 更好的压缩效果。相对于HTTP压缩，Websocket在适当的扩展支持下，可以沿用之前内容的上下文，在传递类似的数据时，可以显著地提高压缩率。

2、环境搭配

2.1、工程创建

---

这里的话狗子我创建的是一个简单的SpringBoot工程，内含了Lombok、fastjson等基础依赖，这一块大家自行发挥即可。

2.2、依赖导入

---

这里需要导入一个Spring家对WebSocket的依赖包，方便后续的配置。

<dependency>
    <groupId>org.springframeworkgroupId>
    <artifactId>spring-websocketartifactId>
    <version>5.3.9version>
dependency>

2.3、配置类

---

通过配置类往容器中注入ServerEndpointExporter从而开启WebSocket支持

@Configuration
public class WebSocketConfig {

    @Bean
    public ServerEndpointExporter serverEndpointExporter()
    {
        return new ServerEndpointExporter();
    }
}

3、具体实现

3.1、前置知识

---

在WebSocket中存在各种触发事件，在Java中这些事件对应着不同的注解，从而对不同事件的逻辑进行自定义操作。

注意：下方提及到的IM为即时通讯Instant Messaging。

3.2、数据封装

---

在WebSocket中主要通过Session进行通讯，注意的是这里的Session并不是我们平时所说的RequestSession，而是javax.websocket包下的Session。而同时还需要一个通讯标识标识不同人之间的通讯连接。如：

• A和B连接通讯的通讯标识为abcd；
• A和C连接通讯的通讯标识为qwer；
• B和A连接通讯的通讯标识为abcd；

@Data
public class WebSocketData {
    /**
     * 当前连接
     */
    private Session session;
    /**
     * 当前通讯ID
     */
    private String communicationId;
}

3.3、思路分析

---

3.4、服务构建

---

在项目中创建server包，并在该包下新建IMServer.java文件，用于编写具体IM的逻辑，其中包含以下属性配合实现具体的通讯逻辑。

@RestController
@Slf4j
@ServerEndpoint(value = "/im/{senderId}/{communicationId}")
public class IMServer {
    /**
     * 记录在线连接
     */
    public static final Map<String, WebSocketData> sessionMap = new ConcurrentHashMap<>();

    /**
     * 封装一个sessionData的类，内含通讯标识和session对象
     */
    private final WebSocketData webSocketData = new WebSocketData();

    /**
     * 用于后续信息存入数据库中使用
     */
    private static MessageService messageService;
    
    @Autowired
    public void setMessageService(MessageService messageService) {
        IMServer.messageService = messageService;
    }
    
    /**
     * @description 获取当前在线人数
     * @method getOnlineCount
     * @author xbaozi
     * @date 2022/9/26 17:11
     **/
    private static synchronized int getOnlineCount() {
        return sessionMap.size();
    }
}

3.5、连接建立

---

/**
 * @param session  与某个客户端的连接会话
 * @param senderId 建立连接的用户
 * @description 连接建立成功调用的方法
 * @method onOpen
 * @author xbaozi
 * @date 2022/9/25 23:05
 **/
@OnOpen
public void onOpen(Session session, @PathParam("senderId") String senderId, @PathParam("communicationId") String communicationId) {
    webSocketData.setSession(session);
    webSocketData.setCommunicationId(communicationId);
    sessionMap.put(senderId, webSocketData);
    log.info("{}: 开始创建连接，新用户{}加入，当前在线人数{}", session, senderId, getOnlineCount());
}

3.6、消息通讯

---

/** 
 * @description 收到客户端消息后调用的方法
 * @method onMessage
 * @author xbaozi 
 * @date 2022/9/27 16:25 
 * @param message   收到的消息
 * @param senderId  发送者ID
 **/
@OnMessage
public void onMessage(String message, @PathParam("senderId") String senderId) {
    WebSocketData senderWebSocketData = sessionMap.get(senderId);
    // 判断发送者连接
    if (senderWebSocketData.getSession() != null) {
        log.info("收到的消息为{}", message);
        // 数据处理
        Message messageObject = JSON.parseObject(message, Message.class);
        log.info("转换成message实体为{}", messageObject);
        try {
            // 发送消息
            sendMessage(senderWebSocketData, messageObject);
        } catch (IOException e) {
            log.info("发送失败，传输出现问题");
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    } else {
        log.info("发送失败，未找到用户userId={}的session", senderId);
    }

}

/**
 * @param senderWebSocketData 内部封装了当前连接信息
 * @param messageObject       需要发送的消息实体
 * @description 发送数据
 * @method sendMessage
 * @author xbaozi
 * @date 2022/9/27 16:22
 **/
private void sendMessage(WebSocketData senderWebSocketData, Message messageObject) throws IOException {
    Session senderSession = senderWebSocketData.getSession();
    if (messageObject == null) {
        // 数据异常，提醒发送者
        senderSession.getBasicRemote().sendText(JSON.toJSONString(Result.error(MISSING_REQUEST_PARAM)));
    } else {
        // 数据写入数据库
        messageObject.setMessageId(UUID.randomUUID().toString(true));
        boolean isSuccess = messageService.save(messageObject);
        if (BooleanUtil.isFalse(isSuccess)) {
            messageObject.setMessageStatus(MESSAGE_STATUS_EXCEPTION);
        }
        // 尝试获取接收者连接
        WebSocketData receiverWebSocketData = sessionMap.get(messageObject.getMessageReceiver());
        // 接收者处于在线状态
        if (receiverWebSocketData.getSession() != null) {
            // 判断当前连接是否为当前通话创建的连接，设置消息为已读
            if (receiverWebSocketData.getCommunicationId().equals(senderWebSocketData.getCommunicationId())) {
                messageObject.setMessageReadFlag(HAS_BEEN_READ);
            }
        }
        // 更新通讯时间
        communicationService.update(new LambdaUpdateWrapper<Communication>()
                .set(Communication::getCommunicationUpdateTime, new Date())
                .eq(Communication::getCommunicationId, senderWebSocketData.getCommunicationId())
        );
        senderSession.getBasicRemote().sendText(JSONUtil.toJsonStr(messageObject));
    }
}

3.7、连接关闭

---

/**
 * @description 连接关闭调用的方法
 * @method onClose
 * @author xbaozi
 * @date 2022/9/25 23:10
 **/
@OnClose
public void onClose(@PathParam("senderId") String senderId) {
    sessionMap.remove(senderId);
    log.info("用户{}连接断开，在线用户{}人……", senderId, getOnlineCount());
}

3.8、连接异常

---

/**
 * @param exception 捕获到的异常
 * @description 发生错误时调用
 * @method onError
 * @author xbaozi
 * @date 2022/9/25 23:11
 **/
@OnError
public void onError(Throwable exception) {
    log.info("发起连接发生了不可描述的错误……");
    exception.printStackTrace();
}

4、结果演示

由于狗子我只是个卑微的后端小菜鸡，所以就没有写前端代码进行测试。这里感谢李士伟开源的demo工程小程序聊天，内含了小程序端的代码，因此这里直接对代码进行修改引用了我的WebSocket服务。在这里我还跑了工程中的后端项目，用作获取联系人等功能，主要展示的就是发送消息和接受到的消息。