• Java之消息推送浅入浅出

    在日常开发中，消息推送是非常典型的业务需求，下面对消息推送简单的分析一下。
    消息推送通常指网站的运营人员通过某种工具对用户当前网页或移动设备APP进行的主动消息推送。主要分为web端消息推送和移动端消息推送。

    • web端消息推送场景：站内信、未读邮件、监控报警；
    • 移动端消息推送场景：有人关注我的公众号，这时我就会收到一条推送消息，以此来吸引我点击打开应用；

    在这里插入图片描述
    在这里插入图片描述
    消息推送通常在服务端会有若张消息推送表，用来记录用户触发不同事件所推送不同类型的消息，前端主动查询或者被动接收用户所有未读的消息。那么消息推送无非是推push和拉pull两种形式。下面看一下具体的实现方式。
    1、短轮询
    轮询应该是实现消息推送方案中最简单的一种。短轮询是指定时间间隔，由浏览器向服务器发出http请求，服务器实时返回未读消息给浏览器，浏览器渲染显示。一个简单的js定时器就可以搞定，每秒钟请求服务器一次未读消息接口，将返回数据显示即可。

    setInterval(() => {
  // 方法请求
  message().then((res) => {
      if (res.code === 200) {
          this.messageData = res.data
      }
  })
}, 1000);

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8

    短轮询实现简单，效果还可以，缺点是由于推送数据并不会频繁变更，无论后端此时是否有新的消息产生，浏览器都会进行请求，势必对服务端造成很大的压力，浪费带宽和服务器资源。
    2、长轮询
    长轮询是对短轮询的一种改进，在尽可能减少对服务器资源的浪费的同时保证消息的相对实时性。如Nacos、Kafka、RocketMQ中都有用到长轮询。
    实例：用长轮询实现消息推送
    因为一个ID可能会被多个长轮询请求监听，所以采用了guava提供的Multimap结构存放长轮询，一个key可以对应多个value。
    如页面发起长轮询请求，请求被挂起，若页面得到请求超时的状态码，则再次发起长轮询请求；当一旦监听到key发生变化，对应的所有长轮询都会得到响应（如数据变更的状态码），则主动请求未读消息的数据接口，渲染页面数据。

    @RestController
@RequestMapping("/polling")
public class PollingController {

    // 存放监听某个Id的长轮询集合
    public static Multimap> watchRequests = Multimaps.synchronizedMultimap(HashMultimap.create());

    /**
     * 设置监听
     */
    @GetMapping(path = "watch/{id}")
    public DeferredResult watch(@PathVariable String id) {
        // 对延迟对象设置超时时间
        DeferredResult deferredResult = new DeferredResult<>(60 * 1000L);
        // 当延迟对象的异步请求完成时移除 key，防止内存溢出
        deferredResult.onCompletion(() -> {
            watchRequests.remove(id, deferredResult);
        });
        // 注册长轮询请求
        watchRequests.put(id, deferredResult);
        return deferredResult;
    }

    /**
     * 变更数据
     */
    @GetMapping(path = "publish/{id}")
    public String publish(@PathVariable String id) {
        // 数据变更 取出监听ID的所有长轮询请求，并一一响应处理
        if (watchRequests.containsKey(id)) {
            Collection> deferredResults = watchRequests.get(id);
            for (DeferredResult deferredResult : deferredResults) {
                deferredResult.setResult("数据已变更：200" + new Date());
            }
        }
        return "success";
    }

}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    @RestControllerAdvice
public class AsyncRequestTimeoutHandler {

    //@ResponseStatus(HttpStatus.NOT_MODIFIED)
    @ExceptionHandler(AsyncRequestTimeoutException.class)
    public String asyncRequestTimeoutHandler(AsyncRequestTimeoutException e) {
        System.out.println("异步请求超时：304" + new Date());
        return "异步请求超时：304" + new Date();
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10

    当请求超过设置的超时时间，会抛出AsyncRequestTimeoutException异常，这里直接用@ControllerAdvice全局捕获统一返回即可，浏览器获取超时状态码后再次发起长轮询请求，如此往复调用。
    长轮询相对于短轮询在性能上提升了，但依然会产生很多的请求，还是有不完美的地方。
    3、SSE
    SSE（Server-sent events），一种服务器发送事件。SSE它是基于HTTP协议的，我们知道一般意义上的HTTP协议是无法做到服务端主动向客户端推送消息的，但SSE是个例外，它变换了一种思路。
    SSE在服务器和客户端之间打开一个单向通道，服务端响应的不再是一次性的数据包而是text/event-stream类型的数据流信息，在有数据变更时从服务器流式传输到客户端。
    整体的实现思路有点类似于在线视频播放，视频流会连续不断的推送到浏览器，你也可以理解成，客户端在完成一次用时很长（网络不畅）的下载。
    4、MOTT
    MQTT（Message Queue Telemetry Transport），一种基于发布/订阅（publish/subscribe）模式的轻量级通讯协议，通过订阅相应的主题来获取消息，是物联网（Internet of Thing）中的一个标准传输协议。
    该协议将消息的发布者（publisher）与订阅者（subscriber）进行分离，因此可以在不可靠的网络环境中，为远程连接的设备提供可靠的消息服务，使用方式与传统的MQ有点类似。
    TCP协议位于传输层，MQTT 协议位于应用层，MQTT 协议构建于TCP/IP协议上，也就是说只要支持TCP/IP协议栈的地方，都可以使用MQTT协议。
    MQTT主要应用在物联网场景，首先HTTP协议它是一种同步协议，客户端请求后需要等待服务器的响应。而在物联网（IOT）环境中，设备会很受制于环境的影响，比如带宽低、网络延迟高、网络通信不稳定等，显然异步消息协议更为适合物联网应用程序。
    HTTP是单向的，如果要获取消息客户端必须发起连接，而在物联网（IOT）应用程序中，设备或传感器往往都是客户端，这意味着它们无法被动地接收来自网络的命令。
    5、Websocket
    WebSocket是一种在TCP连接上进行全双工通信的协议，建立客户端和服务器之间的通信渠道。浏览器和服务器仅需一次握手，两者之间就直接可以创建持久性的连接，并进行双向数据传输。
    实例：springboot+websocket

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

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8

    配置类

    @Configuration
@EnableWebSocket
public class WebSocketConfig {
    /**
     * 这个bean会自动注册使用了@ServerEndpoint注解声明的Websocket endpoint
     */
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11

    websocket服务类

    @Slf4j
@Component
@ServerEndpoint("/websocket/{userId}")
public class WebSocketServer {
    /**
     * 与客户端的连接会话，需要通过它来给客户端发送数据
     */
    private Session session;
    /**
     * 存放客户端容器
     */
    private static final CopyOnWriteArraySet webSockets = new CopyOnWriteArraySet<>();
    /**
     * 用来存在线连接数
     */
    private static final Map sessionPool = new HashMap();
    /**
     * 链接成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session, @PathParam(value = "userId") String userId) {
        try {
            this.session = session;
            webSockets.add(this);
            sessionPool.put(userId, session);
            log.info("websocket消息: 有新的连接，总数为:" + webSockets.size());
        } catch (Exception e) {
        }
    }
    /**
     * 收到客户端消息后调用的方法
     */
    @OnMessage
    public void onMessage(String message) {
        log.info("websocket消息: 收到客户端消息:" + message);
    }
    /**
     * 此为单点消息
     */
    public void sendOneMessage(String userId, String message) {
        Session session = sessionPool.get(userId);
        if (session != null && session.isOpen()) {
            try {
                log.info("websocket消: 单点消息:" + message);
                session.getAsyncRemote().sendText(message);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
    /**
     * 此为单点消息(多人)
     */
    public void sendMoreMessage(String[] userIds, String message) {
        for (String userId : userIds) {
            Session session = sessionPool.get(userId);
            if (session != null && session.isOpen()) {
                try {
                    log.info("【websocket服务端】 单点消息:" + message);
                    session.getAsyncRemote().sendText(message);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67

    测试服务端推送消息给前端

    @CrossOrigin("*")
@RestController
@RequestMapping("/socket")
public class WebsocketController {

    @Resource
    private WebSocketServer webSocketServer;

    /**
     * 发送消息
     */
    @GetMapping(path = "publish")
    public String publish(String message, String userId) {
        //创建业务消息信息
        webSocketServer.sendOneMessage(userId, message);
        return "success";
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18

    页面html

    




    未读消息
    
    
    

    





    
    
      
        
    •  未读消息 0
      • 

    
    
    
    
        
        
        
     接收到的websocket消息
    
        
    
        
    
		
    
        
    
    
    

    




    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73

    测试效果如下
    在这里插入图片描述
    页面初始化建立websocket连接，浏览器和服务端就可以进行双向通信了。
    在实际开发中，需要根据具体的业务场景选择消息推送的方式，需要思考自身业务的特点和场景需求，加油吧，少年。

  • 相关阅读:
    压缩感知常用的重建算法
    计算机中的逻辑运算详解（与、或、非、同或、异或）
    Windows也能拥有好用的命令行吗？Powershell+Terminal折腾记录（v1.0版本）
    Spring Boot中的 6 种API请求参数读取方式
    Java命令行形式将程序打包成jar包,防止报错：没有主清单属性
    支持向量机
    来自jackson的灵魂一击：@ControllerAdvice就能保证万无一失吗？
    MySQL常见问题汇总
    react之Component存在的2个问题
    【Mysql】 linux | mysql | 配置日志路径 | 关闭binlog
  • 原文地址：https://blog.csdn.net/leijie0322/article/details/126139074