SpringBoot整合WebSocket实现订阅消息推送

一、什么是WebSocket
- 1.HTTP协议
- 2.WebSocket协议
二、WebSocket使用场景
- 1.消息推送
- 2.实时聊天
- 3.弹幕
- 4.实时数据更新
三、SpringBoot整合WebSocket（以实现消息推送为例）
- 1.添加依赖
- 2.创建消息类
- 2.WebSocket配置类
- 3.工具类
- 4.测试连接
- 5.服务调用

一、什么是WebSocket

1.HTTP协议

Http协议是一种无状态的超文本传输协议，由客户端向服务器发送请求，服务器向客户端发出响应，特点是一问一答的形式，如果需要服务端主动向客户端发送消息就不适用了。
在这里插入图片描述

2.WebSocket协议

WebSocket是一种计算机通信协议，通过单个 TCP 连接提供双向数据传输的协议。

双向性： 使用 WebSocket 客户端或服务器可以发起消息发送，不必像Http协议，只能由客户端发起请求，因此WebSocket适合服务器单方向客户端发起推送的业务场景。
轻量级： 与 http 相比，WebSocket 消息数据交换要轻得多，因此，在WebSocket客户端和WebSocket服务器进行频繁双向通信时，可以使服务器避免打开多个HTTP连接进行工作来节约资源，提高了工作效率和资源利用率。
数据类型： WebSocket传输支持文本类型、二进制类型（图片、动画等），还可以传输自定义的消息类型。
单个 TCP 连接： 初始连接使用 HTTP，然后将此连接升级到基于套接字的连接。然后这个单一连接用于所有未来的通信。

在这里插入图片描述

二、WebSocket使用场景

HTTP 不适用于两个应用程序之间的长时间实时工通信，但是WebbSocket允许服务端主动向客户端推送数据，并且不用遵循一问一答的规则，具有良好的互动性和时效性。

1.消息推送

公告、新闻推送

2.实时聊天

对话框实时通信

3.弹幕

4.实时数据更新

股票报价、体育比赛分数刷新

三、SpringBoot整合WebSocket（以实现消息推送为例）

来源：spring官方网站给予使用websocket的demo：https://spring.io/guides/gs/messaging-stomp-websocket/

这个项目demo是官网给的示例demo，直接pull下来复制一下就完全可以用了，以下是我把示例demo集成到自己项目的过程：

1.添加依赖

<!-- Websocket -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>

2.创建消息类

就是你想要的消息是怎么样的，有哪些字段和值，例如我的消息类是这样设计的

{
	# 用户连接ID
	"clientId":"36人7829",
	# 消息编码：可进行自定义，例如001是消息通告，002是广告
	"messageCode":"001",
	# 消息内容
	"messageContent":"好消息，打折了",
}

package com.cloud.springnacos9001.entity;

import lombok.AllArgsConstructor;
import lombok.Data;

/**
 * @author caicai
 */
@Data
@AllArgsConstructor
public class WebSocketMessage {

    /**
     * 用户ID
     */
    private String clientId;

    /**
     * 消息编码：可进行自定义，例如001是消息通告，002是广告
     */
    private String messageCode;

    /**
     * 消息内容
     */
    private String messageContent;
}

2.WebSocket配置类

package com.cloud.springnacos9001.config;

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

/**
 * WebSocket配置类
 */
@Configuration
@EnableWebSocket
public class WebSocketConfig {

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

}

3.工具类

package com.cloud.springnacos9001.controller;

import com.cloud.springnacos9001.entity.WebSocketMessage;
import lombok.extern.slf4j.Slf4j;

import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.util.concurrent.ConcurrentHashMap;


/**
 *
 * @author caicai
 * @date
 * @ServerEndpoint
 */
@Component
@Slf4j
@ServerEndpoint(value = "/websocket")
public class WebSocketServer {
    /**
     * 某个客户端连接的session
     */
    private Session session;

    private static ConcurrentHashMap<String,Session> clientMap = new ConcurrentHashMap<>();

    /**
     * 建立连接后调用的方法
     * @param session
     */
    @OnOpen
    public void onOpen(Session session){
        log.info("-----websocket成功连接------");
        this.session = session;
        clientMap.put(this.session.getId(),this.session);
    }

    /**
     * 处理客户端失败连接
     */
    @OnClose
    public void onClose(){
        clientMap.remove(this.session.getId());
        log.info("客户端失去连接,其sessionID为："+this.session.getId());
    }

    /**
     * 连接报错时的处理方式
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error("消息发生错误："+session.getId());
        error.printStackTrace();
    }

    /**
     * 如果需要服务器接受客户端的消息就在这里写了
     * @param text
     */
    @OnMessage
    public void onMessage(String text){
        log.info("接受到客户端传来的消息："+text);
    }

    /**
     * 推送消息
     * getAsyncRemote获取异步端点(推荐使用这个)
     * getBasicRemote获取阻塞端点（不推荐，如果第一次发送失败，就会一直阻塞，影响后面的进行）
     * @param message
     */
    public static void sendMessage(WebSocketMessage message){
        // 遍历向所有用户发送消息
        clientMap.entrySet().stream().forEach(item ->{
            item.getValue().getAsyncRemote().sendText(message.getMessageContent());
        });
    }

    /**
     * 将消息发送给特定的用户
     */
    public static void sendToSpecClient(String clientId,WebSocketMessage webSocketMessage){
        if(clientMap.contains(clientId)){
            clientMap.get(clientId).getAsyncRemote().sendText(webSocketMessage.getMessageContent());
        }else{
            clientMap.remove(clientId);
            // throw new RuntimeException("特定用户消息发生错误，请稍后重试");
        }
    }


}

4.测试连接

用于测试ws的网站：http://www.jsons.cn/websocket/
本机联调的时候可以用
在这里插入图片描述

5.服务调用

 @PostMapping("message")
 public String sendWebSocketMessage(@Param("message") String message){
     String content = "向所有用户推送消息:"+message;
     WebSocketServer.sendMessage(new WebSocketMessage(null,"001",content));
     return "SUCCESS";
 }