SpringBoot2.x集成WebSocket实现广播和点对点推送消息

 WebSocket是一种在单个TCP连接上进行全双工通信的协议。WebSocket 协议在2008年诞生,2011年成为国际标准。现在所有浏览器都已经支持了。它的最大特点就是,服务器可以主动向客户端推送信息,客户端也可以主动向服务器发送信息,是真正的双向平等对话,属于服务器推送技术的一种。

 最近有需求使用WebSocket作为中间件去转发数据,于是就使用SpringBoot2.x与其进行整合,在此期间遇到了一些问题,总算是都解决了。话不多说,直接上代码。

1、添加jar支持,build.gradle如下:

plugins {
    id 'org.springframework.boot' version '2.1.3.RELEASE'
    id 'java'
    id 'idea'
   // id 'war'
}

apply plugin: 'io.spring.dependency-management'

group = 'com.tiantian'
version = '1.0.0'
sourceCompatibility = '1.8'
targetCompatibility = '1.8'

repositories {
    mavenCentral()
}

dependencies {
    testImplementation (
            'org.springframework.boot:spring-boot-starter-test'
    )

    implementation (
            'org.springframework.boot:spring-boot-starter-web',
            'org.springframework.boot:spring-boot-starter-websocket',
            //'org.yeauty:netty-websocket-spring-boot-starter:0.7.4',
            'org.springframework.boot:spring-boot-starter-thymeleaf',
            'org.apache.commons:commons-lang3:3.8.1',
            'com.alibaba:fastjson:1.2.56'
    )

    // lombok支持,打包时处理
    compileOnly ('org.projectlombok:lombok:1.18.6')
    annotationProcessor 'org.projectlombok:lombok:1.18.6'
}

// 使用bootJar打jar包
jar {
    baseName = 'webserver'
    version = '1.0.0'
    manifest {
        attributes "Manifest-Version": 1.0,
                'Main-Class': 'com.tiantian.webserver.WebserverApplication'
    }
}

// 解决代码中的中文编译报错
tasks.withType(JavaCompile) {
    options.encoding = "UTF-8"
}

说明:此项目是打成jar运行的,当然也可以打成war放在Tomcat上运行。

2、配置开启WebSocket功能

/**
 * WebSocket配置
 *
 * @author yueli.liao
 * @date 2019-03-12 11:25
 */
@Configuration
public class WebSocketConfig {

    /**
     * 开启WebSocket功能
     *
     * @return
     */
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
    
}

3、WebSocket核心服务

/**
 * WebSocket服务入口
 *
 * @author yueli.liao
 * @date 2019-03-12 15:16
 */
@Slf4j
@Component
@ServerEndpoint("/websocket/{id}")
public class WebSocketServer {

    // 客户端ID
    private String id = "";

    // 与某个客户端的连接会话,需要通过它来给客户端发送数据
    private Session session;

    // 记录当前在线连接数(为保证线程安全,须对使用此变量的方法加lock或synchronized)
    private static int onlineCount = 0;

    // 用来存储当前在线的客户端(此map线程安全)
    private static ConcurrentHashMap<String, WebSocketServer> webSocketMap = new ConcurrentHashMap<>();

    /**
     * 连接建立成功后调用
     */
    @OnOpen
    public void onOpen(@PathParam(value = "id") String id, Session session) {
        this.session = session;
        this.id = id; // 接收到发送消息的客户端编号
        webSocketMap.put(id, this); // 加入map中
        addOnlineCount();           // 在线数加1
        log.info("客户端" + id + "加入,当前在线数为:" + getOnlineCount());
        try {
            sendMessage("WebSocket连接成功");
        } catch (IOException e) {
            log.error("WebSocket IO异常");
        }
    }

    /**
     * 连接关闭时调用
     */
    @OnClose
    public void onClose() {
        webSocketMap.remove(this); // 从map中删除
        subOnlineCount();               // 在线数减1
        log.info("有一连接关闭,当前在线数为:" + getOnlineCount());
    }

    /**
     * 收到客户端消息后调用
     *
     * @param message 客户端发送过来的消息<br/>
     *                消息格式:内容 - 表示群发,内容|X - 表示发给id为X的客户端
     * @param session
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        log.info("来自客户端的消息:" + message);
        String[] messages = message.split("[|]");
        try {
            if (messages.length > 1) {
                sendToUser(messages[0], messages[1]);
            } else {
                sendToAll(messages[0]);
            }
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
    }

    /**
     * 发生错误时回调
     *
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error("WebSocket发生错误");
        error.printStackTrace();
    }

    /**
     * 推送信息给指定ID客户端,如客户端不在线,则返回不在线信息给自己
     *
     * @param message 客户端发来的消息
     * @param sendClientId 客户端ID
     * @throws IOException
     */
    public void sendToUser(String message, String sendClientId) throws IOException {
        if (webSocketMap.get(sendClientId) != null) {
            if (!id.equals(sendClientId)) {
                webSocketMap.get(sendClientId).sendMessage("客户端" + id + "发来消息:" + " <br/> " + message);
            } else {
                webSocketMap.get(sendClientId).sendMessage(message);
            }
        } else {
            // 如客户端不在线,则返回不在线信息给自己
            sendToUser("当前客户端不在线", id);
        }
    }

    /**
     * 推送发送信息给所有人
     *
     * @param message 要推送的消息
     * @throws IOException
     */
    public void sendToAll(String message) throws IOException {
        for (String key : webSocketMap.keySet()) {
            webSocketMap.get(key).sendMessage(message);
        }
    }

    /**
     * 推送消息
     *
     * @param message 要推送的消息
     * @throws IOException
     */
    private void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
    }

    private static synchronized int getOnlineCount() {
        return onlineCount;
    }

    private static synchronized void addOnlineCount() {
        WebSocketServer.onlineCount++;
    }

    private static synchronized void subOnlineCount() {
        WebSocketServer.onlineCount--;
    }

}

4、为解决大文本(超过WebSocket协议限制的长度)不能发送的问题,有两种解决办法,一种是在Tomcat \conf\web.xml文件添加如下代码:

<!-- 注:单位为byte -->
<context-param>
    <param-name>org.apache.tomcat.websocket.textBufferSize</param-name>
    <param-value>1024000</param-value>
</context-param>
<context-param>
    <param-name>org.apache.tomcat.websocket.binaryBufferSize</param-name>
    <param-value>1024000</param-value>
</context-param>

另一种是在服务启动时动态进行设置,在项目中添加如下代码:

/**
 * 解决WebSocket传输内容过长的问题
 * 注:若发送内容过长时,服务端无内容输出,连接会自动关闭
 *
 * @author yueli.liao
 * @date 2019-03-13 13:35
 */
@Slf4j
@Configuration
@ComponentScan
@EnableAutoConfiguration
public class WebAppRootContext implements ServletContextInitializer {

    @Value("${websocket.bufferSize}")
    private String bufferSize;

    @Override
    public void onStartup(ServletContext servletContext) throws ServletException {
        log.info("WebSocket最大文本长度:" + bufferSize);
        servletContext.addListener(WebAppRootListener.class);
        servletContext.setInitParameter("org.apache.tomcat.websocket.textBufferSize", bufferSize);
    }

}

代码中的websocket.bufferSize,在application.properties进行配置,如下:

#websocket textSize(单位为byte)
websocket.bufferSize=1024000

说明:推荐使用第二种方式,这样可移植性强,对部署和开发都很方便。

5、测试,也有两种方式,采用哪种或两种都可以。

  • 一是在Google或Firefox浏览器中安装WebSocket Client插件进行测试;
  • 二是在项目中写相关的JS进行测试,代码如下:
<!DOCTYPE HTML>
<html>
<head>
    <title>WebSocket测试</title>
</head>

<body>
Welcome<br/>
<input id="text" type="text" />
<button onclick="send()">发送消息</button>
<button onclick="closeWebSocket()">关闭连接</button>
<div id="message">
</div>
</body>

<script type="text/javascript">
    var websocket = null;

    // 判断当前浏览器是否支持WebSocket
    if('WebSocket' in window){
        // 为了方便测试,故将链接写死
        websocket = new WebSocket("ws://localhost:8088/websocket/1");
    } else{
        alert('Not support websocket')
    }

    // 连接发生错误的回调方法
    websocket.onerror = function(){
        setMessageInnerHTML("发生错误");
    };

    // 连接成功建立的回调方法
    websocket.onopen = function(event){
        setMessageInnerHTML("打开连接");
    }

    // 接收到消息的回调方法
    websocket.onmessage = function(event){
        setMessageInnerHTML(event.data);
    }

    // 连接关闭的回调方法
    websocket.onclose = function(){
        setMessageInnerHTML("关闭连接");
    }

    // 监听窗口关闭事件,当窗口关闭时,主动去关闭websocket连接,防止连接还没断开就关闭窗口,server端会抛异常
    window.onbeforeunload = function(){
        websocket.close();
    }

    // 将消息显示在网页上
    function setMessageInnerHTML(innerHTML){
        document.getElementById('message').innerHTML += innerHTML + '<br/>';
    }

    // 关闭连接
    function closeWebSocket(){
        websocket.close();
    }

    // 发送消息
    function send(){
        var message = document.getElementById('text').value;
        websocket.send(message);
    }
</script>
</html>

说明:此webserver可以进行点对点推送消息,也可以进行广播。源代码已上传到GitHub,可自行下载。

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 205,386评论 6 479
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 87,939评论 2 381
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 151,851评论 0 341
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 54,953评论 1 278
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 63,971评论 5 369
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,784评论 1 283
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 38,126评论 3 399
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,765评论 0 258
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 43,148评论 1 300
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,744评论 2 323
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,858评论 1 333
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,479评论 4 322
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 39,080评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,053评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,278评论 1 260
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 45,245评论 2 352
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,590评论 2 343

推荐阅读更多精彩内容