springboot2整合WebSocket以和STOMP

04.18为什么使用WebSocket

WebSocket是一种协议，基于TCP的一种新的网络协议，它实现了浏览器与服务器全双工通信，允许服务器主动发送信息给客户端。那么为什么要去使用webSocket呢，因为普遍的web项目或者http协议都是由客服端发起，后端响应数据并返回。但是这样会存在一个问题，如果后台的数据刷新了，而前端还在查看旧的数据。这时一定要重新由客户端发起才能达到刷新数据的效果。

我参考了该博主的帖子很详细的说明了websocket的demo制作过程。

使用WebSocket步骤

1.导入Maven依赖

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

2.添加配置文件

@Configuration //开启WebSocket支持
public class WebSocketConfig {
    @Bean
    public ServerEndpointExporter serverEndpointExporter(){
        return new ServerEndpointExporter();
    }
}

3.WebSocketServer

WebSocketServer是类似客户端服务单的形式(使用ws协议),其实也相当于是一个controller。使用注解@@ServerEndpoint("/imserver/{userId}") 、@Component启用即可。连接之后使用OnOpen开启连接，前端也要点击调用之后加入到socket中。开启就意味着也要关闭连接，在刷新页面或者关闭页面时，就调用OnClose来关闭连接。使用onMessage来接收信息，sendMessage发送信息。

新建一个ConcurrentHashMap webSocketMap 用于接收当前userId的WebSocket，方便IM之间对userId进行推送消息。单机版本的实现就基本完成了，如果要使用集群版就要涉及到sql或者redis等缓存工具存储和处理了。并改造对应的sendMessage方法。

/**
 * 因为ServerEndpoint涉及到多例模式，所以要是用this，将数据更新或发出*/
@ServerEndpoint("/imserver/{userId}")
@Component
public class WebSocketServer {
    /**记录日志 使用的是*/
    static Log log = LogFactory.get(WebSocketServer.class);
    /**声明静态变量，用来记录当前在线连接数，设计成线程安全。*/
    private static int onlineCount = 0;
    /**concurrent包的线程安全，用来存放每个客户端对应的MyWebSocket对象。*/
    private static ConcurrentHashMap<String,WebSocketServer> webSocketMap = new ConcurrentHashMap<>();
    /**与某个客户端的连接会话，需要通过它来给客户端发送数据*/
    private Session session;
    /**
     * 接收userId
     */
    private String userId="";

声明有关在线人数的方法，使用同步锁，这样就保证只会有一个进程在操作人数。有人open 就记录人数加一，有人close人数就减一，只要检测到人数的变化，就在日志中输出。

public static synchronized int getOnlineCount() {
    return onlineCount; //记录在线人数
}

public static synchronized void addOnlineCount() {
    WebSocketServer.onlineCount  ; //每次调用 人数加一
}

public static synchronized void subOnlineCount() {
    WebSocketServer.onlineCount--; //每次调用 人数减一
}

通过@OnOpen开启。因为websocket涉及到多例，也就是多个线程调用此方法，所以使用this关键字来确保是同一个客户端在操作。传入的id如果已经包含了，就从map中移除，重新加入新的。反之则添加在线人数。并打印日志，当前在线人数。

@OnOpen
public void OnOpen(Session session, @PathParam("userId")String userId){
    this.session = session;
    this.userId = userId;
    if(webSocketMap.containsKey(userId)){
        webSocketMap.remove(userId);
        //加入map中
        webSocketMap.put(userId,this);
    }else{
        //加入map中
        webSocketMap.put(userId,this);
        //在线数加1
        addOnlineCount();
    }
    log.info("用户连接:" userId ",当前在线人数为:"   getOnlineCount());

    try {
        sendMessage("连接成功");
    } catch (IOException e) {
        log.error("用户:" userId ",网络异常!!!!!!");
    }
}

关闭时调用的方法，从map从移除id。并打印日志。

@OnClose
public void onClose() {
    if(webSocketMap.containsKey(userId)){
        webSocketMap.remove(userId);
        //从map中删除
        subOnlineCount();
    }
    log.info("用户退出:" userId ",当前在线人数为:"   getOnlineCount());
}

收到客户端的消息后调用的方法，首先打印日志，将谁发起的id和发出的信息打印。然后判断发送的消息是否不为空。不为空的话就使用fastjson解析数据。将信息转换成object类型，(此时message包含接收人和具体信息)并且把发送人也添加到object中，最后继续判断接收人的id是否是在线的，在线的才进行发送信息的请求，不在线就返回日志不在该服务器上。最后抛一下异常。

@OnMessage
public void onMessage(String message, Session session) {
    log.info("用户消息:" userId ",报文:" message);
    //可以群发消息
    //消息保存到数据库、redis
    if(StringUtils.isNotBlank(message)){
        try {
            //解析发送的报文
            JSONObject jsonObject = JSON.parseObject(message);
            //追加发送人(防止串改)
            jsonObject.put("fromUserId",this.userId);
            String toUserId=jsonObject.getString("toUserId");
            //传送给对应toUserId用户的websocket
            if(StringUtils.isNotBlank(toUserId)&&webSocketMap.containsKey(toUserId)){
 webSocketMap.get(toUserId).sendMessage(jsonObject.toJSONString());
            }else{
                log.error("请求的userId:" toUserId "不在该服务器上");
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

然后也会存在断网，连接不上的情况，我们就要调用OnError方法，在日志中打印

@OnError
public void onError(Session session, Throwable error) {
    log.error("用户错误:" this.userId ",原因:" error.getMessage());
    error.printStackTrace();
}

发送信息的方法，就是onmessage或者连接时给客户端提醒，控制台打印。

public void sendMessage(String message) throws IOException {
    this.session.getBasicRemote().sendText(message);
}

最后比较关键的是后台发向前台的信息，也是核心功能之一，虽然实现的只是简单的一对一的模式，通过id,信息往前台发送信息，消息直接呈现到客户端口。

/**
 * 发送自定义消息  后台->前台
 * */
public static void sendInfo(String message, @PathParam("userId") String userId) throws IOException {
    log.info("发送消息到:" userId "，讯息:" message);
    if(StringUtils.isNotBlank(userId)&&webSocketMap.containsKey(userId)){
        webSocketMap.get(userId).sendMessage(message);
    }else{
        log.error("用户" userId ",不在线！");
    }
}

4.DemoController

最后简单的使用控制层做个请求

@RestController
public class DemoController {
	
    @GetMapping("index")
    public ResponseEntity<String> index(){
        return ResponseEntity.ok("请求成功");
    }
	//请求到主页，这个是使用freemarker解析ModelAndView，生成的页面
    @GetMapping("page")
    public ModelAndView page(){
        return new ModelAndView("websocket");
    }
	//后台->前台信息的请求
    @RequestMapping("/push/{toUserId}")
    public ResponseEntity<String> pushToWeb(String message, @PathVariable("toUserId")String toUserId) throws IOException {
        WebSocketServer.sendInfo(message,toUserId);
        return ResponseEntity.ok("MSG SEND SUCCESS");
    }
}

5.前台页面

<!DOCTYPE html>
<html>
<head>
    <meta charset="utf-8">
    <title>websocket通讯</title>
</head>
<script src="https://cdn.bootcss.com/jquery/3.3.1/jquery.js"></script>
<script>
    var socket;
    function openSocket() {
        if(typeof(WebSocket) == "undefined") {
            console.log("您的浏览器不支持WebSocket");
        }else{
            console.log("您的浏览器支持WebSocket");
            //实现化WebSocket对象，指定要连接的服务器地址与端口  建立连接
            //等同于socket = new WebSocket("ws://localhost:8080/xxxx/im/25");
            //var socketUrl="${request.contextPath}/im/" $("#userId").val();
            //var socketUrl="http://localhost:8080/demo/imserver/" $("#userId").val();
            var socketUrl="http://localhost:8080/demo/imserver/" $("#userId").val();
            socketUrl=socketUrl.replace("https","ws").replace("http","ws");
            console.log(socketUrl);
            if(socket!=null){
                socket.close();
                socket=null;
            }
            socket = new WebSocket(socketUrl);
            //打开事件
            socket.onopen = function() {
                console.log("websocket已打开");
                //socket.send("这是来自客户端的消息"   location.href   new Date());
            };
            //获得消息事件
            socket.onmessage = function(msg) {
                console.log(msg.data);
                //发现消息进入    开始处理前端触发逻辑
            };
            //关闭事件
            socket.onclose = function() {
                console.log("websocket已关闭");
            };
            //发生了错误事件
            socket.onerror = function() {
                console.log("websocket发生了错误");
            }
        }
    }
    function sendMessage() {
        if(typeof(WebSocket) == "undefined") {
            console.log("您的浏览器不支持WebSocket");
        }else {
            console.log("您的浏览器支持WebSocket");
            console.log('{"toUserId":"' $("#toUserId").val() '","contentText":"' $("#contentText").val() '"}');
            socket.send('{"toUserId":"' $("#toUserId").val() '","contentText":"' $("#contentText").val() '"}');
        }
    }
</script>
<body>
<p>【userId】：<div><input id="userId" name="userId" type="text" value="10"></div>
<p>【toUserId】：<div><input id="toUserId" name="toUserId" type="text" value="20"></div>
<p>【toUserId】：<div><input id="contentText" name="contentText" type="text" value="hello websocket"></div>
<p>【操作】：<div><a onclick="openSocket()">开启socket</a></div>
<p>【操作】：<div><a onclick="sendMessage()">发送消息</a></div>
</body>
</html>

6.测试效果

首先开启2个客户端，分别写上两个不同的id之间相互通信，都开启websocket：

我们用id为10的向20的发送一条消息，然后在20的客户端查看收到的信息：

可以看到成功的发起了10与20之间的通信。并且发送之后可以看到两边的id。发送者也会收到一条自己发出的讯息，而接收者则是看到由谁发出，发给谁的信息以及信息内容。最后使用postman测试一下后台发送到客户端的信息

最后看一下idea中控制台日志中的信息：

可以清晰的看到谁发出信息和后台发送信息。至此完成websocket的简单demo。

04.20基于WebSocket实现的stomp发布订阅

首先，STOMP是websocket协议的一种实现方式，Streaming Text Orientated Message Protocol，是流文本定向消息协议。它提供了一个可互操作的连接格式，允许STOMP客服端与任意STOMP消息代理(Broker)进行交互。
我参考了该博主的文章

使用STOMP技术步骤

1.导入依赖，配置文件

之前做过类似的步骤，STOMP也是websocket包下的。只是配置文件略有变化，同样注册成@Configuration，让spring识别成配置类。并且开启messageBroker代理的注解。然后去实现重写broker接口中的方法，最主要的两个，一个是实现订阅的队列，另一个开启endpoint，让前端页面启动时就能通过连接到STOMP端口中。

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
    @Override
    public void configureMessageBroker(MessageBrokerRegistry config){
        // 注册topic实现广播，user实现点对点, alone 点对点，mass群发 也是一种广播
        config.enableSimpleBroker("/topic","/user","/alone","/mass");
    }
    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        //注册两个STOMP的endpoint，分别用于广播和点对点
        //广播类型，后端->前端
        registry.addEndpoint("/publicServer").setAllowedOriginPatterns("*").withSockJS();
        //点对点类型 客户端之间 有单独alone 也有 mass 群发
       //注册STOMP的endpoint，并指定使用SockJS协议，原使用setAllowedOrigin，但根据不同的boot版本，改用setAllowedOriginPatterns，都是用来解决跨域的问题 前端->前端
        registry.addEndpoint("/privateServer").setAllowedOriginPatterns("*").withSockJS();
    }
}

2.创建实体类

使用了2个实体类，一个用来接收发送请求的页面信息，另一个接收返回给前端页面。

@Data
public class ChatRoomRequest {
    private String userId;
    private String name;
    private String chatValue;
}
@Data
public class ChatRoomResponse {
    private String userId;
    private String name;
    private String chatValue;
}

3.ChatRoomController

制作一个controller专门使用springboot封装好的SimpMessagingTemplate来调用convertAndSendToUser()实现往指定的userId,指定的订阅队列，传递的消息。这样的形式发送信息。当然，还有特殊的注解@MessageMapping,意思就是该消息发送到哪个订阅队列中。与@SendTo效果一样，如果一起使用，信息会传递两次。

至于为什么mass中不用实例中的方法，因为发出的讯息大家都看的到，不用加上限制条件。

@Controller
public class ChatRoomController {
    @Autowired
    public SimpMessagingTemplate template;

    @GetMapping("/chatroom")
    public String  chatroom (){
        return "chatroom";
    }
    //群发
    //@MessageMapping("/mass")  向/mass这个订阅路径发送信息，与sendTo /mass效果一致。
    //SendTo 发送至 Broker 下的指定订阅路径
    @SendTo("/mass")
    public ChatRoomResponse mass(ChatRoomRequest chatRoomRequest){
        //方法用于群发测试
        System.out.println("name = "   chatRoomRequest.getName());
        System.out.println("chatValue = "   chatRoomRequest.getChatValue());
        ChatRoomResponse response=new ChatRoomResponse();
        response.setName(chatRoomRequest.getName());
        response.setChatValue(chatRoomRequest.getChatValue());
        return response;
    }
    //单独聊天
    @MessageMapping("/alone")
    public ChatRoomResponse alone(ChatRoomRequest chatRoomRequest){
        //方法用于一对一测试
        System.out.println("userId = "   chatRoomRequest.getUserId());
        System.out.println("name = "   chatRoomRequest.getName());
        System.out.println("chatValue = "   chatRoomRequest.getChatValue());
        ChatRoomResponse response=new ChatRoomResponse();
        //response.setUserId(chatRoomRequest.getUserId());
        response.setName(chatRoomRequest.getName());
        response.setChatValue(chatRoomRequest.getChatValue());
        this.template.convertAndSendToUser(chatRoomRequest.getUserId() "","/alone",response);
        return response;
    }
}

4.前端页面(重点)

(2改版)，添加了取消订阅的功能。

因为使用的STOMP.js触发STOMP服务，所以要在前端启用stompClient，连接到上述我们定义好的两种endpoint其中一个。然后订阅相应的队列，再调用对应的controller中的方法完成交互。再一个就是，启动页面之后，自动订阅/mass，相当于加入群聊。这样大家发送的消息，只要是连接了STOMP的都能收到群聊中的消息。然后就是选择完角色之后会默认先订阅自己的ID下的/alone。去私聊其他人的时候就会往对应ID下的/alone队列发送消息，这样就能保证你发送的消息，对应ID(select中的选项)下的人能收到消息。

<!DOCTYPE html>
<html lang="zh-CN" xmlns:th="http://www.thymeleaf.org">
<head>
    <title>群聊版</title>
    <link rel="stylesheet" type="text/css" th:href="@{/css/chatroom.css}">
    <script  th:src="@{https://cdn.bootcss.com/sockjs-client/1.1.4/sockjs.min.js}"></script>
    <script th:src="@{https://cdn.bootcss.com/stomp.js/2.3.3/stomp.min.js}"></script>
    <script th:src="@{https://code.jquery.com/jquery-3.2.0.min.js}"
            integrity="sha256-JAW99MJVpJBGcbzEuXk4Az05s/XyDdBomFqNlM3ic I=" crossorigin="anonymous"></script>
</head>

<body>
<div>
    <div style="float:left;width:40%">
        <p>请选择你是谁：</p>
        <select id="selectName" onchange="sendAloneUser();">
            <option value="1">请选择</option>
            <option value="aa">AA</option>
            <option value="bb">BB</option>
            <option value="cc">CC</option>
            <option value="dd">DD</option>
            <option value="ee">EE</option>
        </select>
        <div class="chatWindow">
            <p style="color:darkgrey">群聊：</p>
            <section id="chatRecord" class="chatRecord">
                <p id="titleval" style="color:#CD2626;"></p>
            </section>
            <section class="sendWindow">
                <textarea name="sendChatValue" id="sendChatValue" class="sendChatValue"></textarea>
                <input type="button" name="sendMessage" id="sendMessage" class="sendMessage" onclick="sendMassMessage()" value="发送">
            </section>
        </div>
        <div style="margin-left:60px;margin-top:5px;">
            <input type="button" name="unSubscribe" onclick="unSubscribe()" value="退出群聊">
        </div>
    </div>


    <div style="float:right; width:40%">
        <p>请选择你要发给谁：</p>
        <select id="selectName2">
            <option value="1">请选择</option>
            <option value="aa">AA</option>
            <option value="bb">BB</option>
            <option value="cc">CC</option>
            <option value="dd">DD</option>
            <option value="ee">EE</option>
        </select>
        <div class="chatWindow">
            <p style="color:darkgrey">单独聊：</p>
            <section id="chatRecord2" class="chatRecord">
                <p id="titleval" style="color:#CD2626;"></p>
            </section>
            <section class="sendWindow">
                <textarea name="sendChatValue2" id="sendChatValue2" class="sendChatValue"></textarea>
                <input type="button" name="sendMessage" id="sendMessage" class="sendMessage" onclick="sendAloneMessage()" value="发送">
            </section>
        </div>
    </div>
</div>

<script>
    var stompClient = null;

    //加载完浏览器后  调用connect（），打开双通道
    $(function(){
        //打开双通道
        connect()
    })

    //强制关闭浏览器  调用websocket.close（）,进行正常关闭
    window.onunload = function() {
        disconnect()
    }

    //打开双通道
    function connect(){
        var socket = new SockJS('http://localhost:8080/privateServer'); //连接SockJS的endpoint名称为"privateServer"
        stompClient = Stomp.over(socket);//使用STMOP子协议的WebSocket客户端
        stompClient.heartbeat.outgoing = 20000;
        // client will send heartbeats every 20000ms
        stompClient.heartbeat.incoming = 0;
        stompClient.connect({},function(frame){//连接WebSocket服务端

        console.log('Connected:'   frame);
        //广播接收信息
        stompTopic();

        });
    }

    //关闭双通道
    function disconnect(){
        if(stompClient != null) {
            stompClient.disconnect();
        }
        console.log("Disconnected");
    }

    //广播（一对多）选择自己是谁的时候就订阅了mass
    function stompTopic(){
        //通过stompClient.subscribe订阅/mass(群聊) 目标(destination)发送的消息（广播接收信息）
        stompClient.subscribe('/mass',function(response){
        var message=JSON.parse(response.body);
        //展示广播的接收的内容接收
        var response = $("#chatRecord");
        response.append("<p><span>" message.name ":</span><span>" message.chatValue "</span></p>");
        });
    }

    //列队（一对一）
    function stompQueue(){

        var userId=$("#selectName").val();
        alert("监听:" userId)
        //通过stompClient.subscribe订阅/user下 具体id的 目标(destination)发送的消息（队列接收信息）
        stompClient.subscribe('/user/'   userId   '/alone',function(response){
        var message=JSON.parse(response.body);
        //展示一对一的接收的内容接收
        var response = $("#chatRecord2");
        response.append("<p><span>" message.name ":</span><span>" message.chatValue "</span></p>");
        });
    }

    //选择发送给谁的时候触发连接服务器
    function sendAloneUser(){
        stompQueue();
    }

    //退出群聊，退订/mass
    function unSubscribe(){
        stompClient.unsubscribe('sub-0');
        console.log("取消订阅成功 -> destination: /mass");
    }

    //群发
    function sendMassMessage(){
        var postValue={};
        var chatValue=$("#sendChatValue");
        var userName=$("#selectName").val();
        postValue.name=userName;
        postValue.chatValue=chatValue.val();
        if(userName==1||userName==null){
            alert("请选择你是谁！");
            return;
        }
        if(chatValue==""||userName==null){
            alert("不能发送空消息！");
            return;
        }
        stompClient.send("/mass",{},JSON.stringify(postValue));
        chatValue.val("");
    }

    //单独发 私聊
    function sendAloneMessage(){
        var postValue={};
        var chatValue=$("#sendChatValue2");
        var userName=$("#selectName").val();
        var sendToId=$("#selectName2").val();
        var response = $("#chatRecord2");
        postValue.name=userName;
        postValue.chatValue=chatValue.val();
        postValue.userId=sendToId;
        if(userName==1||userName==null){
            alert("请选择你是谁！");
            return;
        }
        if(sendToId==1||sendToId==null){
            alert("请选择你要发给谁！");
            return;
        }
        if(chatValue==""||userName==null){
            alert("不能发送空消息！");
            return;
        }
        stompClient.send("/alone",{},JSON.stringify(postValue));
        response.append("<p><span>" userName ":</span><span>" chatValue.val() "</span></p>");
        chatValue.val("");
    }
</script>
</body>
</html>

5.测试效果

首先开启3个客户端，分别选好各自的角色：

可以看到启动页面之后，默认订阅了/mass。这样就会接收到群聊中的消息了。然后选完角色之后又会订阅自己的/alone

我们再测试一下群聊和私聊的效果

私聊效果

使用AA角色分别给BB和CC发布一条消息

然后切换到BB角色查看消息，可以看到发布给CC的消息，BB不会收到

这里可能会问，明明是发布到ID为bb，cc的/alone队列中，为什么自己也能看到呢？因为在触发调用之前，会先往自己的版块中发送同样的消息，这样也能识别自己发过的消息。

群聊效果

其实从私聊的效果图中就可以看到我在角色A发送的消息，B也收到了消息。因为A调用的controller是往/mass队列中发布消息，只要订阅了/mass的用户都能看到消息。

取消订阅

我们使用角色AA先发布一条消息，然后取消订阅，然后用其他角色发布群聊消息，回到A中看群聊部分

可以看到取消订阅后，的确收不到广播中的消息了。但是我采用的是死办法，在取消订阅中直接使用了/mass中的id取消订阅，由于我暂时还没吃透stomp中的函数，暂时先实现效果，后面再动态获取退订的id。

	//退出群聊，退订/mass
    function unSubscribe(){
        stompClient.unsubscribe('sub-0');
        console.log("取消订阅成功 -> destination: /mass");
    }

这篇好文章是转载于：学新通技术网