电商项目之Web实时消息推送（附源码）

在这里插入图片描述

文章目录

  • 1 问题背景
  • 2 前言
  • 3 什么是消息推送
  • 4 短轮询
  • 5 长轮询
    • 5.1 demo代码
  • 6 iframe流
    • 6.1 demo代码
  • 7 SSE
    • 7.1 demo代码
    • 7.2 生产环境的应用 （重要）
  • 8 MQTT

1 问题背景

扩宽自己的知识广度，研究一下web实时消息推送

2 前言

  1. 文章参考自Web 实时消息推送的 7 种实现方案
  2. 针对一些比较重要的方式，我都会尽量敲出一份完整的demo代码，享受其中的编程乐趣。
  3. 在SSE方式中，笔者延申思考，将他应用于电商支付的场景中，给出了比较合理的解决方案，但并未在生产环境中验证，仍待考证。

3 什么是消息推送

消息推送是指服务端将消息推送给客户端。常见的场景有：有人关注公众号，公众号推送消息给关注者；站内消息通知；未读邮件数量；监控告警数量等等。

4 短轮询

常见的http请求即是短轮询，由客户端发起请求，服务端接收请求并同步实时处理，最后返回数据给客户端。

5 长轮询

短轮询的异步方式即是长轮询，异步在哪里？客户端发起请求，web容器（比如tomcat）安排子线程去处理这些请求，将这些请求交给服务端后，无需阻塞等待结果，tomcat会立即安排该子线程理其他请求 ，tomcat以此接收更多的请求提升系统的吞吐量。服务端处理完请求再返回数据给客户端。

5.1 demo代码

因为一个ID可能会被多个长轮询请求监听，所以采用了guava包提供的Multimap结构存放长轮询，一个key可以对应多个value。一旦监听到key发生变化，对应的所有长轮询都会响应。

引入guava依赖

<dependency><groupId>com.google.guava</groupId><artifactId>guava</artifactId><version>31.1-jre</version>
</dependency>

处理请求的接口：

package com.ganzalang.gmall.sse.controller;import com.google.common.collect.HashMultimap;
import com.google.common.collect.Multimap;
import com.google.common.collect.Multimaps;
import org.springframework.stereotype.Controller;
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.ResponseBody;
import org.springframework.web.context.request.async.DeferredResult;import java.time.LocalDateTime;
import java.util.Collection;
import java.util.Date;@Controller
@RequestMapping("/polling")
public class PollingController {/*** 关于 DeferredResult 还有一个很重要的点：请求的处理线程（即 tomcat 线程池的线程）不会等到 DeferredResult#setResult() 被调用才释放，而是直接释放了。* 而 DeferredResult 的做法就类似仅把事情安排好，不会管事情做好没，tomcat 线程就释放走了，注意此时不会给请求方（如浏览器）任何响应，而是将请求存放在一边，* 咱先不管它，等后面有结果了再把之前的请求拿来，把值响应给请求方。*/// 存放监听某个Id的长轮询集合// 线程同步结构public static Multimap<String, DeferredResult<String>> watchRequests = Multimaps.synchronizedMultimap(HashMultimap.create());public static final long TIME_OUT = 100000;/*** 设置监听*/@GetMapping(path = "watch/{id}")@ResponseBodypublic DeferredResult<String> watch(@PathVariable String id) {// 延迟对象设置超时时间DeferredResult<String> deferredResult = new DeferredResult<>(TIME_OUT);// 异步请求完成时移除 key，防止内存溢出deferredResult.onCompletion(() -> {watchRequests.remove(id, deferredResult);});// 注册长轮询请求watchRequests.put(id, deferredResult);return deferredResult;}/*** 变更数据*/@GetMapping(path = "publish/{id}")@ResponseBodypublic String publish(@PathVariable String id) {// 数据变更 取出监听ID的所有长轮询请求，并一一响应处理if (watchRequests.containsKey(id)) {Collection<DeferredResult<String>> deferredResults = watchRequests.get(id);for (DeferredResult<String> deferredResult : deferredResults) {deferredResult.setResult("我更新了" + LocalDateTime.now());}}return "success";}/*** 监听器的数量*/@GetMapping(path = "listener/num")@ResponseBodypublic int num() {return watchRequests.size();}
}

当请求超过设置的超时时间，会抛出AsyncRequestTimeoutException异常，这里直接用@ControllerAdvice全局捕获统一返回即可，前端获取约定好的状态码后再次发起长轮询请求，如此往复调用。代码如下：

package com.ganzalang.gmall.sse.exception.handler;import org.springframework.http.HttpStatus;
import org.springframework.web.bind.annotation.ControllerAdvice;
import org.springframework.web.bind.annotation.ExceptionHandler;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.ResponseStatus;
import org.springframework.web.context.request.async.AsyncRequestTimeoutException;@ControllerAdvice
public class AsyncRequestTimeoutHandler {@ResponseStatus(HttpStatus.NOT_MODIFIED)@ResponseBody@ExceptionHandler(AsyncRequestTimeoutException.class)public String asyncRequestTimeoutHandler(AsyncRequestTimeoutException e) {System.out.println("异步请求超时");return "304";}
}

测试：

首先页面发起长轮询请求/polling/watch/10086监听消息更变，请求被挂起，不变更数据直至超时，再次发起了长轮询请求；紧接着手动变更数据/polling/publish/10086，长轮询得到响应，前端处理业务逻辑完成后再次发起请求，如此循环往复。

6 iframe流

在页面中插入一个隐藏的<iframe>标签，通过在src中请求消息数量API接口，由此在服务端和客户端之间创建一条长连接，服务端持续向iframe传输数据。传输的数据通常是html，js脚本。

6.1 demo代码

笔者打算使用一个页面来展示效果，因此需要引入一个freemarker依赖

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-freemarker</artifactId>
</dependency>

yml中配置freemarker：

spring:freemarker:suffix: .ftlcontent-type: text/htmlcharset: UTF-8cache: false# ftl页面存放的路径template-loader-path: classpath:/templates/

写一个ftl页面：

在这里插入图片描述

ftl页面源码如下：

<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><title>IFRAME</title>
</head>
<body>
<iframe src="/iframe/message" style="display:none"></iframe>
<div><h1>clock</h1><span id="clock"></span><h1>count</h1><span id="count"></span>
</div>
</body>
</html>

服务端的代码：

package com.ganzalang.gmall.sse.controller;import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.time.LocalDateTime;
import java.util.concurrent.atomic.AtomicInteger;@Controller
@RequestMapping("/iframe")
@Slf4j
public class IframeController {/*** 访问首页** @param request* @return* @throws IOException*/@GetMapping("/index")public String index(HttpServletRequest request) throws IOException {log.info("iframe-index");return "iframe-index";}/*** 返回消息** @param response* @throws IOException* @throws InterruptedException*/@GetMapping(path = "message")public void message(HttpServletResponse response) throws IOException, InterruptedException {AtomicInteger count = new AtomicInteger(1);while (true) {log.info("current time:{}", LocalDateTime.now());response.setHeader("Pragma", "no-cache");response.setDateHeader("Expires", 0);response.setHeader("Cache-Control", "no-cache,no-store");response.setStatus(HttpServletResponse.SC_OK);response.getWriter().print(" <script type=\"text/javascript\">\n" + "parent.document.getElementById('clock').innerHTML = \"" + count.get() + "\";" + "parent.document.getElementById('count').innerHTML = \"" + count.getAndIncrement() + "\";" + "</script>");}}
}

测试：

访问http://localhost:8033/iframe/index即可，大家会发现这样非常占用服务器资源，服务端会很卡。并且客户端还会一直在loading，如下图所示：

在这里插入图片描述

7 SSE

Server-sent events，简称SSE。SSE在服务器和客户端之间打开一个单向通道，服务端响应的不再是一次性的数据包而是text/event-stream类型的数据流信息，在有数据变更时从服务器流式传输到客户端。SSE有如下几个特点：

  1. 基于HTTP
  2. 单向通信
  3. 实现简单，无需引入其组件
  4. 默认支持断线重连 （服务端重启，客户端会重新发送连接请求，这是天生解决服务端发版的问题啊）
  5. 只能传送文本信息

7.1 demo代码

SSE同样是使用页面展示效果，需要添加一些freemarker相关东西，具体细节可见第6.1节

页面代码：

在这里插入图片描述

页面源码：

<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><title>IFRAME</title>
</head>
<body>
<div id="message"></div>
<script>let source = null;// 获取url中userId参数的值。如url=http://localhost:8033/sse/index?userId=1111var userId = window.location.search.substring(1).split('&')[0].split('=')[1];// 判断当前客户端（浏览器）是否支持SSE，有些浏览器不是默认支持SSE的if (window.EventSource) {// 建立连接source = new EventSource('http://localhost:8033/sse/sub/'+userId);document.getElementById("message").innerHTML += "连接用户=" + userId + "<br>";/*** 连接一旦建立，就会触发open事件* 另一种写法：source.onopen = function (event) {}*/source.addEventListener('open', function (e) {document.getElementById("message").innerHTML += "建立连接。。。<br>";}, false);/*** 客户端收到服务器发来的数据* 另一种写法：source.onmessage = function (event) {}*/source.addEventListener('message', function (e) {document.getElementById("message").innerHTML += e.data + "<br>";});} else {document.getElementById("message").innerHTML += "你的浏览器不支持SSE<br>";}
</script></body>
</html>

服务端代码：

package com.ganzalang.gmall.sse.controller;import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Controller;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;@Controller
@RequestMapping("/sse")
@Slf4j
public class SseController {private static Map<String, SseEmitter> sseEmitterMap = new ConcurrentHashMap<>();/*** 推送消息给客户端** @param userId* @param msg* @throws IOException* @throws InterruptedException*/@GetMapping(path = "/message/{userId}")public void message(@PathVariable("userId") String userId, @RequestParam(value = "msg", required = false)String msg) throws IOException, InterruptedException {String message = StringUtils.isEmpty(msg) ? "pay success" : msg;sendMessage(userId, message);}/*** 查询当前的sse连接数量** @return* @throws IOException* @throws InterruptedException*/@GetMapping(path = "/num")@ResponseBodypublic String num() throws IOException, InterruptedException {return String.valueOf(sseEmitterMap.keySet().size());}@GetMapping(path = "/del/{userId}")@ResponseBodypublic String num(@PathVariable("userId") String userId) throws IOException, InterruptedException {sseEmitterMap.remove(userId);return "success";}/*** 开启sse连接** @param userId* @return* @throws IOException* @throws InterruptedException*/@GetMapping("/sub/{userId}")@ResponseBodypublic SseEmitter sub(@PathVariable("userId") String userId) throws IOException, InterruptedException {SseEmitter sseEmitter = connect(userId);log.info("userId={}, result:{}", userId, "Pay success");return sseEmitter;}/*** 访问sse首页** @return* @throws IOException*/@GetMapping("/index")public String index() throws IOException {log.info("sse-index");return "sse-index";}/*** 创建连接** @date: 2022/7/12 14:51*/public static SseEmitter connect(String userId) {try {// 设置超时时间，0表示不过期。默认30秒SseEmitter sseEmitter = new SseEmitter(0L);// 注册回调sseEmitter.onCompletion(() -> removeUser(userId));sseEmitter.onError((e) -> log.error("exception:{}", e.getMessage(), e));sseEmitter.onTimeout(() -> removeUser(userId));sseEmitterMap.put(userId, sseEmitter);return sseEmitter;} catch (Exception e) {log.info("创建新的sse连接异常，当前用户：{}", userId);}return null;}/*** 给指定用户发送消息** @date: 2022/7/12 14:51*/public static void sendMessage(String userId, String message) {if (sseEmitterMap.containsKey(userId)) {try {sseEmitterMap.get(userId).send(message);} catch (IOException e) {log.error("用户[{}]推送异常:{}", userId, e.getMessage());removeUser(userId);}}}/*** 移除对应的客户端连接* * @param userId*/private static void removeUser(String userId) {sseEmitterMap.remove(userId);}
}

测试：

访问http://localhost:8033/sse/index?userId=1111注册客户端连接，此处记为index页面。

浏览器另开一个tab页，访问http://localhost:8033/sse/message/1111?msg=haha，然后去index页面看，会发现有消息展示出来了：

在这里插入图片描述

当服务端重启，客户端会自动重连，即index页面的那个http请求会再次发给服务端

7.2 生产环境的应用 （重要）

笔者做支付比较多，该sse方式也可以用于做支付结果的消息通知（一般都是用短轮询做查询，查询支付结果；那么现在可以使用SSE方式）。针对应用于生产环境，笔者认为有如下几点需要注意：

  1. 由前面服务端代码可见，服务端需要在内存中保存客户端的连接（那个sseEmitterMap）。在服务端是集群的情况下，接收客户端请求的服务端节点的内存中，并不一定就有客户端的连接，此处可以使用Redis的发布订阅功能，通知存有客户端连接的服务端节点进行发消息。除了Redis发布订阅，还能通过Redis+RPC做一个精准调用，Redis可以存储Map<客户端连接的唯一标识, 服务端节点IP>，拿到IP后通过RPC进行精准调用，详情可以见服务端实时推送技术之SSE（Server-Send Events）。

  2. 服务端节点中的每个客户端连接都需要做超时处理，超时则将连接从内存中移除，否则会发生OOM。

  3. 假如服务端发版，内存中的所有客户端连接都会丢失，但无需担忧，因为客户端默认会重连。

8 MQTT

MQTT方式需要借助消息队列来实现，其实相当于常规的生产者消费者模式。因实现起来比较复杂，（需要搭建MQ），笔者此处暂不研究MQTT具体实现。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/234991.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

用户注册这样玩，保你平安

用户注册这样玩，保你平安

前言 基本上每个系统系统都包含用户注册、发送验证码等基本操作。在前些年&#xff0c;我还记得我在逛 csdn、贴吧、网易新闻等网站的时候是可以不登陆也能浏览完网页内容的&#xff0c;但是近几年这些网站已经改成了不登陆不让用&#xff0c;浏览网页时不时提醒你要进行登录&…
阅读更多...
二分算法（整数二分、浮点数二分）

二分算法（整数二分、浮点数二分）

文章目录 二分一、整数二分&#xff08;一&#xff09;整数二分思路&#xff08;二&#xff09;整数二分算法模板1.左查找&#xff08;寻找左侧边界&#xff09;2.右查找&#xff08;寻找右侧边界&#xff09;3.总模板 &#xff08;三&#xff09;题目&#xff1a;数的范围 二、…
阅读更多...
一键删除方舟编译器缓存文件js、js.map插件ArkCompilerSupport

一键删除方舟编译器缓存文件js、js.map插件ArkCompilerSupport

新手学习鸿蒙开发&#xff0c;发现DevEco Studio编译过种会生成js、js.map&#xff0c;在论坛上看了其它开发者也提了问题但无没解决&#xff0c;写了一个插件大家试下&#xff1a; https://plugins.jetbrains.com/plugin/23192-arkcompilersupport 源码&#xff1a;https://g…
阅读更多...
咨询＋低代码，强强联合为制造业客户赋能

咨询＋低代码，强强联合为制造业客户赋能

内容来自演讲&#xff1a;沈毅 | 遨睿智库 | 董事长 & 王劭禹 | 橙木智能 | 联合创始人 摘要 文章主要讲述了智库董事长沈毅创办广告公司的经历&#xff0c;以及他在管理公司过程中遇到的问题和挑战&#xff0c;最后通过与明道云以及橙木智能联合创始人王邵禹老师的合作&…
阅读更多...
Ubuntu systemd-analyze命令（系统启动性能分析工具：分析系统启动时间，找出可能导致启动缓慢的原因）

Ubuntu systemd-analyze命令（系统启动性能分析工具：分析系统启动时间，找出可能导致启动缓慢的原因）

文章目录 Ubuntu systemd-analyze命令剖析目录简介systemd与systemd-analyze工作原理 安装和使用命令参数详解用例与示例显示启动时间&#xff08;systemd-analyze time&#xff09;列出启动过程中各个服务的启动时间&#xff08;systemd-analyze blame&#xff09;显示系统启动…
阅读更多...
使用JDBC操作数据库时，插入数据中文乱码

使用JDBC操作数据库时，插入数据中文乱码

如图&#xff1a; 解决办法&#xff1a; 修改连接数据库的路径&#xff0c;即url 如下&#xff1a; 设置编码格式为utf-8 urljdbc:mysql://localhost:3306/qfedu?useUnicodetrue&characterEncodingUTF-8再次运行&#xff0c;插入数据即可
阅读更多...
XTU OJ 1339 Interprime 学习笔记

XTU OJ 1339 Interprime 学习笔记

链接 传送门 代码 #include<bits/stdc.h> using namespace std;const int N1e610; //78498 我计算了一下&#xff0c;6个0的范围内有这么多个素数&#xff0c;所以开这么大的数组存素数 //计算的代码是一个循环 int prime[80000]; int a[N],s[N];//s数组是前缀和数组b…
阅读更多...
vscode Markdown 预览样式美化多方案推荐

vscode Markdown 预览样式美化多方案推荐

优雅的使用 vscode写 Markdown&#xff0c;预览样式美化 1 介绍 我已经习惯使用 vscode 写 markdown。不是很喜欢他的 markdown 样式&#xff0c;尤其是代码块高亮的样式。当然用 vscode 大家基本上都会选择安装一个Markdown-preview-enhanced的插件&#xff0c;这个插件的确…
阅读更多...
【DDD】领域驱动设计总结——如何构造领域模型

【DDD】领域驱动设计总结——如何构造领域模型

文章目录 一 分离领域二 领域对象分类2.1 实体(ENTITY)2.2 值对象(VALUE OBJECT)2.3 服务(SERVICE)2.4 模块&#xff08;&#xff2d;ODULE&#xff09; 三 管理领域对象的生命周期3.1 聚合&#xff08;AGGREGATE&#xff09;3.2 工厂&#xff08;FACTORY&#xff09;3.3 存储库…
阅读更多...
mysql：免费的GUI客户端工具推荐并介绍常用的操作

mysql：免费的GUI客户端工具推荐并介绍常用的操作

给大家推荐几个常用的 mysql 数据库客户端 sequel-pro sequel-ace 官网下载地址 免费 sequel-ace 可以理解为 Sequel Pro 的升级版&#xff0c;由于Sequel Pro官方不维护了&#xff0c;特别是对 MySQL 8.0 支持不好&#xff0c;所以现在由社区维护了新分支 sequel-ace&#x…
阅读更多...
具备这四个特征的项目经理，牛逼！

具备这四个特征的项目经理，牛逼！

大家好&#xff0c;我是老原。 成为一个业绩第一又能准时下班的工作强人&#xff0c;应该是每个职场人的梦想&#xff0c;但现实往往不那么尽如人意…… 虽然如此&#xff0c;但是不代表我们不能向能做到这样的大佬看齐啊。 工作十余年&#xff0c;见过各种各样的职场人士&a…
阅读更多...
C++——解锁string常用接口

C++——解锁string常用接口

目录 string::npos; 1.测试string容量相关的接口&#xff1a; 1.1 string::size() 1.2 string::clear() 1.3 string::resize() 1.4 string::erase() 1.5 string::reserve() 保留 1.6 std::string::shrink_to_fit 2.string数据插入删除相关的接口 2.1 std::string::pus…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023