[转] SpringBoot2.0集成WebSocket,实现后台向前端推送信息

2021-02-14 22:19

阅读:400

YPE html>

标签:ping   数据库   关于   contex   val   运行   研究   key   error   

什么是WebSocket?

技术图片

 

 WebSocket协议是基于TCP的一种新的网络协议。它实现了浏览器与服务器全双工(full-duplex)通信——允许服务器主动发送信息给客户端。

为什么需要 WebSocket?
初次接触 WebSocket 的人,都会问同样的问题:我们已经有了 HTTP 协议,为什么还需要另一个协议?它能带来什么好处?

答案很简单,因为 HTTP 协议有一个缺陷:通信只能由客户端发起,HTTP 协议做不到服务器主动向客户端推送信息。

技术图片

 

 举例来说,我们想要查询当前的排队情况,只能是页面轮询向服务器发出请求,服务器返回查询结果。轮询的效率低,非常浪费资源(因为必须不停连接,或者 HTTP 连接始终打开)。因此WebSocket 就是这样发明的。

前言
以下教程更新于2020-01-05 :

整合了IM相关的优化
优化开启/关闭连接的处理
上传到开源项目spring-cloud-study-websocket,方便大家下载代码。
感谢大家的支持和留言,14W访问量是满满的动力!接下来还会有websocket+redis集群优化篇针对多ws服务器做简单优化处理,敬请期待!

话不多说,马上进入干货时刻。

maven依赖
SpringBoot2.0对WebSocket的支持简直太棒了,直接就有包可以引入

        org.springframework.boot  
           spring-boot-starter-websocket  
       

WebSocketConfig

启用WebSocket的支持也是很简单,几句代码搞定

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

/**
 * 开启WebSocket支持
 * @author zhengkai.blog.csdn.net
 */
@Configuration  
public class WebSocketConfig {  
    
    @Bean  
    public ServerEndpointExporter serverEndpointExporter() {  
        return new ServerEndpointExporter();  
    }  
  
} 

WebSocketServer
这就是重点了,核心都在这里。

因为WebSocket是类似客户端服务端的形式(采用ws协议),那么这里的WebSocketServer其实就相当于一个ws协议的Controller

直接@ServerEndpoint("/imserver/{userId}") 、@Component启用即可,然后在里面实现@OnOpen开启连接,@onClose关闭连接,@onMessage接收消息等方法。

新建一个ConcurrentHashMap webSocketMap 用于接收当前userId的WebSocket,方便IM之间对userId进行推送消息。单机版实现到这里就可以。

集群版(多个ws节点)还需要借助mysql或者redis等进行处理,改造对应的sendMessage方法即可。

package com.softdev.system.demo.config;

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 com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang.StringUtils;
import org.springframework.stereotype.Component;
import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;


/**
 * @author zhengkai.blog.csdn.net
 */
@ServerEndpoint("/imserver/{userId}")
@Component
public class WebSocketServer {

    static Log log=LogFactory.get(WebSocketServer.class);
    /**静态变量,用来记录当前在线连接数。应该把它设计成线程安全的。*/
    private static int onlineCount = 0;
    /**concurrent包的线程安全Set,用来存放每个客户端对应的MyWebSocket对象。*/
    private static ConcurrentHashMap 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);
            //加入set中
            addOnlineCount();
            //在线数加1
        }

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

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

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        if(webSocketMap.containsKey(userId)){
            webSocketMap.remove(userId);
            //从set中删除
            subOnlineCount();
        }
        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){
                e.printStackTrace();
            }
        }
    }

    /**
     *
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error("用户错误:"+this.userId+",原因:"+error.getMessage());
        error.printStackTrace();
    }
    /**
     * 实现服务器主动推送
     */
    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 synchronized int getOnlineCount() {
        return onlineCount;
    }

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

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

消息推送

至于推送新信息,可以再自己的Controller写个方法调用WebSocketServer.sendInfo();即可

import com.softdev.system.demo.config.WebSocketServer;
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 java.io.IOException;

/**
 * WebSocketController
 * @author zhengkai.blog.csdn.net
 */
@RestController
public class DemoController {

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

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

    @RequestMapping("/push/{toUserId}")
    public ResponseEntity pushToWeb(String message, @PathVariable String toUserId) throws IOException {
        WebSocketServer.sendInfo(message,toUserId);
        return ResponseEntity.ok("MSG SEND SUCCESS");
    }
}

页面发起

页面用js代码调用websocket,当然,太古老的浏览器是不行的,一般新的浏览器或者谷歌浏览器是没问题的。还有一点,记得协议是ws的,如果使用了一些路径类,可以replace(“http”,“ws”)来替换协议。

websocket通讯

【userId】:

【toUserId】:

【toUserId】:

【操作】:

开启socket

【操作】:

发送消息

运行效果
v20200105,加入开源项目spring-cloud-study-websocket,更新运行效果,更方便理解。
v1.1的效果,刚刚修复了日志,并且支持指定监听某个端口,代码已经全部更新,现在是这样的效果
打开两个页面,按F12调出控控制台查看测试效果:
页面 参数
http://localhost:9999/demo/page fromUserId=10,toUserId=20
http://localhost:9999/demo/page fromUserId=20,toUserId=10
分别开启socket,再发送消息

技术图片

 

 

2. 向前端推送数据:

http://localhost:9999/demo/push/10?message=123123

技术图片

 

 

通过调用push api,可以向指定的userId推送信息,当然报文这里乱写,建议规定好格式。

后续
针对简单IM的业务场景,进行了一些优化,可以看后续的文章SpringBoot2+WebSocket之聊天应用实战(优化版本)(v20201005已整合)

主要变动是CopyOnWriteArraySet改为ConcurrentHashMap,保证多线程安全同时方便利用map.get(userId)进行推送到指定端口。

相比之前的Set,Set遍历是费事且麻烦的事情,而Map的get是简单便捷的,当WebSocket数量大的时候,这个小小的消耗就会聚少成多,影响体验,所以需要优化。在IM的场景下,指定userId进行推送消息更加方便。

Websocker注入Bean问题
关于这个问题,可以看最新发表的这篇文章,在参考和研究了网上一些攻略后,项目已经通过该方法注入成功,大家可以参考。
关于controller调用controller/service调用service/util调用service/websocket中autowired的解决方法

netty-websocket-spring-boot-starter
Springboot2构建基于Netty的高性能Websocket服务器(netty-websocket-spring-boot-starter)
只需要换个starter即可实现高性能websocket,赶紧使用吧

Springboot2+Netty+Websocket
Springboot2+Netty实现Websocket,使用官方的netty-all的包,比原生的websocket更加稳定更加高性能,同等配置情况下可以handle更多的连接。

代码样式全部已经更正,另外也感谢大家的阅读和评论,一起进步,谢谢!~~

serverEndpointExporter错误

org.springframework.beans.factory.BeanCreationException: Error creating bean with name ‘serverEndpointExporter’ defined in class path resource [com/xxx/WebSocketConfig.class]: Invocation of init method failed; nested exception is java.lang.IllegalStateException: javax.websocket.server.ServerContainer not available

 

感谢@来了老弟儿 的反馈:

如果tomcat部署一直报这个错,请移除 WebSocketConfig 中@Bean ServerEndpointExporter 的注入 。

ServerEndpointExporter 是由Spring官方提供的标准实现,用于扫描ServerEndpointConfig配置类和@ServerEndpoint注解实例。使用规则也很简单:

如果使用默认的嵌入式容器 比如Tomcat 则必须手工在上下文提供ServerEndpointExporter。
如果使用外部容器部署war包,则不需要提供提供ServerEndpointExporter,因为此时SpringBoot默认将扫描服务端的行为交给外部容器处理,所以线上部署的时候要把WebSocketConfig中这段注入bean的代码注掉。
————————————————
版权声明:本文为CSDN博主「Moshow郑锴」的原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接及本声明。
原文链接:https://blog.csdn.net/moshowgame/java/article/details/80275084

 

[转] SpringBoot2.0集成WebSocket,实现后台向前端推送信息

标签:ping   数据库   关于   contex   val   运行   研究   key   error   

原文地址:https://www.cnblogs.com/eedc/p/12719021.html


评论


亲,登录后才可以留言!