springboot2 搭建webstock 简单例子

参考：

https://blog.csdn.net/moshowgame/article/details/80275084

需求：

在交易时，后台价格发生变化，实时通知告知前端。

现有方案：

通过http网页主动请求刷新。

目前采用的是网页循环方式调用后台 settimeout。。。目前交互不及时，刷新频繁了，对后台不好，刷新存在时间差。

如果后台有变化了实时通知前台，那么就不会存在无效刷新的情况了。

改造：

使用webstocket，后台有变化，主动推送给前端。因为建立的是长连接。所以，变化通知是相当快的。

场景还有很多，不说了，直接上代码

application配置文件

配置文件

server: 
  port: 10130
spring: 
  application:
    name: ccjr-business-webstocket

#thymeleaf start
#spring.thymeleaf.mode=HTML5
spring.thymeleaf.encoding: UTF-8
spring.thymeleaf.content-type: text/html
#开发时关闭缓存,不然没法看到实时页面
spring.thymeleaf.cache: false
spring.thymeleaf.prefix: classpath:/templates/
spring.thymeleaf.suffix: .html
#thymeleaf end

java配置文件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;
/**
 * @Desc : 注明bean Bean会自动注册使用@ServerEndpoint注解申明的websocket endpoint
 * @author : Chenweixian 陈惟鲜
 * @Date : 2021年2月25日 下午4:36:36
 */
@Configuration
@EnableWebSocket
public class WebSocketConfig {
 
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
    
}

服务类

import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;

import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;

import org.apache.commons.lang.StringUtils;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;

import lombok.extern.slf4j.Slf4j;

/**
 * @Desc : 服务端服务类
 * @author : Chenweixian 陈惟鲜
 * @Date : 2021年2月25日 下午4:33:21
 */
@ServerEndpoint("/imserver/{userId}")
@Component
@Slf4j
public class ImWebSocketServer {

    /** concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。 */
    private static ConcurrentHashMap<String, ImWebSocketServer> webSocketMap = new ConcurrentHashMap<>();
    /** 与某个客户端的连接会话，需要通过它来给客户端发送数据 */
    private Session session;
    /** 接收userId */
    private String userId = "";

    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("userId") String userId) {
        this.session = session;
        this.userId = userId;
        if (webSocketMap.containsKey(userId)) {
            webSocketMap.remove(userId);
            webSocketMap.put(userId, this);
            // 加入set中
        } else {
            webSocketMap.put(userId, this);
        }

        log.info("用户连接:" + userId + ",当前在线人数为:" + getOnlineCount());

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

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        if (webSocketMap.containsKey(userId)) {
            webSocketMap.remove(userId);
        }
        log.info("用户退出:" + userId + ",当前在线人数为:" + getOnlineCount());
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message
     *            客户端发送过来的消息
     */
    @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 + "不在该服务器上");
                    // 否则不在这个服务器上，发送到mysql或者redis
                }
            } catch (Exception e) {
                log.error("异常", e);
            }
        }
    }

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

    /** 在线人数 */
    public static synchronized int getOnlineCount() {
        return webSocketMap.size();
    }
    
    /**
     * 实现服务器主动推送
     */
    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
    }

    /**
     * 发送自定义消息
     */
    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 + ",不在线！");
        }
    }
    
    /**
     * 发送自定义消息--群发
     */
    public static void sendInfoGroup(String message) throws IOException {
        if(ImWebSocketServer.webSocketMap != null && ImWebSocketServer.webSocketMap.size() > 0) {
            for (String userId: ImWebSocketServer.webSocketMap.keySet()) {
                sendInfo(message, userId);
            }
        }
    }
}

controller类 主要页面展示消息推送

package com.ccjr.business.web.controller;

import java.io.IOException;

import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.ModelAndView;

import com.ccjr.business.web.stocket.ImWebSocketServer;

/**
 * 页面跳转
 * WebSocketController
 * @author 陈惟鲜
 */
@RestController
public class IndexController {

    @GetMapping("index")
    public ResponseEntity<String> index(){
        return ResponseEntity.ok("请求成功");
    }

    @GetMapping("page")
    public ModelAndView page(){
        return new ModelAndView("index");
    }

    /**
     * 单独给某用户发信息
     * @author : Chenweixian
     * @Date : 2021年2月25日 下午4:39:54
     * @param message
     * @param toUserId
     * @return
     * @throws IOException
     */
    @RequestMapping("/push/{toUserId}")
    public ResponseEntity<String> pushToWeb(String message, @PathVariable String toUserId) throws IOException {
        ImWebSocketServer.sendInfo(message, toUserId);
        return ResponseEntity.ok("MSG SEND SUCCESS");
    }
    
    /**
     * 发送消息给所有用户
     * @author : Chenweixian
     * @Date : 2021年2月25日 下午4:39:54
     * @param message
     * @param toUserId
     * @return
     * @throws IOException
     */
    @RequestMapping("/pushall")
    public ResponseEntity<String> pushall(String message) throws IOException {
        ImWebSocketServer.sendInfoGroup(message);
        return ResponseEntity.ok("MSG SEND SUCCESS");
    }
}

页面调用 index.html

<!DOCTYPE html>
<html>
<head>
    <meta charset="utf-8">
    <title>im-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:8888/xxxx/im/25");
            //var socketUrl="${request.contextPath}/im/"+$("#userId").val();
            var socketUrl="http://localhost:10130/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已打开");
                setMessageInnerHTML("websocket已打开");
            };
            //获得消息事件
            socket.onmessage = function(msg) {
                // console.log(msg.data);
                setMessageInnerHTML(msg.data);
                //发现消息进入    开始处理前端触发逻辑
            };
            //关闭事件
            socket.onclose = function() {
                // console.log("websocket已关闭");
                setMessageInnerHTML("websocket已关闭");
            };
            //发生了错误事件
            socket.onerror = function() {
                // console.log("websocket发生了错误");
                setMessageInnerHTML("websocket发生了错误");
            }
        }
    }
    function sendMessage() {
        if(typeof(WebSocket) == "undefined") {
            console.log("您的浏览器不支持WebSocket");
            setMessageInnerHTML("您的浏览器不支持WebSocket");
        }else {
            // console.log("您的浏览器支持WebSocket");
            console.log('{"toUserId":"'+$("#toUserId").val()+'","contentText":"'+$("#contentText").val()+'"}');
            setMessageInnerHTML('{"toUserId":"'+$("#toUserId").val()+'","contentText":"'+$("#contentText").val()+'"}');
            socket.send('{"toUserId":"'+$("#toUserId").val()+'","contentText":"'+$("#contentText").val()+'"}');
        }
    }
    //将消息显示在网页上
    function setMessageInnerHTML(innerHTML) {
        document.getElementById('message').innerHTML += innerHTML + '<br/>';
    }
</script>
<body>
<div><p>【userId】：<input id="userId" name="userId" type="text" value="10"></div>
<div><p>【toUserId】：<input id="toUserId" name="toUserId" type="text" value="20"></div>
<div><p>【sendMessage】：<input id="contentText" name="contentText" type="text" value="hello websocket"></div>
<div><p>【操作】：<a onclick="openSocket()">开启socket</a></div>
<div><p>【操作】：<a onclick="sendMessage()">发送消息</a></div>
<div id="message">
</body>

</html>

eclipse 控制台：

测试对话两个网页：

如图：