总阅读量:次

⭐⭐⭐ Spring Boot 项目实战 ⭐⭐⭐ Spring Cloud 项目实战
《Dubbo 实现原理与源码解析 —— 精品合集》 《Netty 实现原理与源码解析 —— 精品合集》
《Spring 实现原理与源码解析 —— 精品合集》 《MyBatis 实现原理与源码解析 —— 精品合集》
《Spring MVC 实现原理与源码解析 —— 精品合集》 《数据库实体设计合集》
《Spring Boot 实现原理与源码解析 —— 精品合集》 《Java 面试题 + Java 学习指南》

摘要: 原创出处 www.jianshu.com/p/62132f605669 「清风徐来水波不清」欢迎转载，保留摘要，谢谢！

🙂🙂🙂关注**微信公众号：【芋道源码】**有福利：

  1. RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表
  2. RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址
  3. 您对于源码的疑问每条留言都将得到认真回复。甚至不知道如何读源码也可以请教噢。
  4. 新的源码解析文章实时收到通知。每周更新一篇左右。
  5. 认真的源码交流微信群。

前言

前面文章也有 websocket 相关的文章，为什么这次又要重新写一篇呢？第一这篇文章需求业务场景有些不同，第二这篇文章 websocket 基本上完全基于注解操作简单。 其实能实现定时消息推送的技术有很多，Dwr、goeasy、comer4j 、netPush 等技术也可以完全实现这个功能.

  • DWR 之前文档的消息推送也有使用到，但是在实际项目中表现的并不是很好，毕竟技术相对较老，对于一些浏览器版本兼容性不是很好，而且容易出现消息丢失的情况，研究半天源码改动很多无法解决这个问题。
  • GoEasy 其实在消息推送方面表现还是比较良好，但是其是收费的，我们项目基本使用开源产品对应收费产品合规性检测肯定无法通过，因此没有考虑。
  • 后面两个技术没有具体研究过，目前觉得 webSocket 可以完美解决我现在业务需求。

websocket 简介

websocket 协议是在 http 协议上的一种补充协议，是 html5 的新特性，是一种持久化的协议。其实 websocket 和 http 关系并不是很大，不过都是属于应用层的协议。关于更多概念大家可以参考下面文章讲述的很详细，接下来我们就开始实战。

https://www.cnblogs.com/fuqiang88/p/5956363.html

websocket 定时推送

本教程基于 springboot 为脚手架，没使用过 springboot 同学可以看往期文章，或者直接去 spring 官网拉一个 springboot 基础项目下来。

加入依赖

在 springboot 的项目中添加一下 webSocket 依赖，一般一项新技术的引入在 springboot 中也只是引用一个此技术 starter 的依赖，其他配置基本 springboot 帮我们解决了。

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

配置

新建一个 Java 配置类，注入 ServerEndpointExporter 配置，如果是使用 springboot 内置的 tomcat 此配置必须，如果是使用的是外部 tomcat 容器此步骤请忽略。看 spring 源码中这样描述，使用此配置可以关闭 servlet 容器对 websocket 端点的扫描，这个暂时没有深入研究。

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

}

核心代码

接下来最核心的类其实就是提供一个前后端交互的类实现消息的接收推送。

  • @ServerEndpoint(value = "/wsdemo") 前端通过此 URI 和后端交互，建立连接
  • @Component 不用说将此类交给 spring 管理
  • @OnOpen websocket 建立连接的注解，前端触发上面 URI 时会进入此注解标注的方法，和之前关于 DWR 文章中的 onpage 方法类似
  • @OnClose 顾名思义关闭连接，销毁 session
  • @OnMessage 收到前端传来的消息后执行的方法

@ServerEndpoint(value = "/wsdemo")
@Component
public class MyWebSocket {
    /**静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。*/
    private static int onlineCount = 0;

    /** concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
        在外部可以获取此连接的所有websocket对象，并能对其触发消息发送功能，我们的定时发送核心功能的实现在与此变量 */
    private static CopyOnWriteArraySet<MyWebSocket> webSocketSet = new CopyOnWriteArraySet<MyWebSocket>();

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

    /**
     * 连接建立成功调用的方法
     * 
     * 类似dwr的onpage方法，参考之前文章中demo有
     * */
    @OnOpen
    public void onOpen(Session session) {
        this.session = session;
        webSocketSet.add(this);     //加入set中
        addOnlineCount();           //在线数加1
        System.out.println("有新连接加入！当前在线人数为" + getOnlineCount());
        try {
            sendMessage("连接已建立成功.");
        } catch (Exception e) {
            System.out.println("IO异常");
        }
    }

    /**
     * 连接关闭调用的方法
     * 
     * 参考dwrsession摧毁方法
     */
    @OnClose
    public void onClose() {
        webSocketSet.remove(this);  //连接关闭后，将此websocket从set中删除
        subOnlineCount();           //在线数减1
        System.out.println("有一连接关闭！当前在线人数为" + getOnlineCount());
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message 客户端发送过来的消息*/
    @OnMessage
    public void onMessage(String message, Session session) {
        System.out.println("来自客户端的消息:" + message);
    }

     // 错误提示
     @OnError
     public void onError(Session session, Throwable error) {
         System.out.println("发生错误");
         error.printStackTrace();
     }

     // 发送消息，在定时任务中会调用此方法           
     public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
 
     }


  

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

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

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



    public Session getSession() {
        return session;
    }

    public void setSession(Session session) {
        this.session = session;
    }

    public static CopyOnWriteArraySet<MyWebSocket> getWebSocketSet() {
        return webSocketSet;
    }

    public static void setWebSocketSet(CopyOnWriteArraySet<MyWebSocket> webSocketSet) {
        MyWebSocket.webSocketSet = webSocketSet;
    }
}

定时任务

使用 spring 的 Schedule 建立定时任务

  • @EnableScheduling 开启 spring 定时任务功能
  • @Scheduled(cron = "0/10 * * * * ?") 用于标识定时执行的方法，此处主要方法返回值一定是 void，没有入参。对应定时时间配置可以百度 cron 语法，根据自己的业务选择合适的周期 在这类中，我们通过上面 MyWebSocket 提供的静态方法获取其中的 webSocketSet ，来获取所有此业务相关的所有 websocketsession，可以在定时任务中对 session 内容进行验证判断（权限验证等），进行发送消息

@Component
@EnableScheduling
public class TimeTask
{
    private static Logger logger = LoggerFactory.getLogger(TimeTask.class);
    @Scheduled(cron = "0/1 * * * * ?")   //每分钟执行一次
    public void test(){
        System.err.println("*********   定时任务执行   **************");
        CopyOnWriteArraySet<MyWebSocket> webSocketSet = 
     MyWebSocket.getWebSocketSet();
            int i = 0 ;
        webSocketSet.forEach(c->{
            try {
                c.sendMessage("  定时发送  " + new Date().toLocaleString());
            } catch (IOException e) {
                e.printStackTrace();
            }
        });

        System.err.println("/n 定时任务完成.......");
    }

}

前端页面

前端页面可以参考使用，主要要更改调用的 url 为自己项目 URL

<!DOCTYPE HTML>
<html>
<head>
    <title>My WebSocket</title>
</head>

<body>
Welcome<br/>
<input id="text" type="text" /><button onclick="send()">Send</button>    <button onclick="closeWebSocket()">Close</button>
<div id="message">
</div>
</body>

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

    //判断当前浏览器是否支持WebSocket  ,主要此处要更换为自己的地址
    if('WebSocket' in window){
        websocket = new WebSocket("ws://localhost:8886/wsdemo");
    }
    else{
        alert('Not support websocket')
    }

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

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

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

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

    //监听窗口关闭事件，当窗口关闭时，主动去关闭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>

效果演示

文章目录
  1. 1. 前言
  2. 2. websocket 简介
  3. 3. websocket 定时推送
    1. 3.1. 加入依赖
    2. 3.2. 配置
    3. 3.3. 核心代码
    4. 3.4. 定时任务
    5. 3.5. 前端页面
  4. 4. 效果演示