上海网站建设口碑最好的公司,如何自学动漫设计,青海建设局网站,flash网站做seo优化好不好文章目录 1 问题背景2 前言3 什么是消息推送4 短轮询5 长轮询5.1 demo代码 6 iframe流6.1 demo代码 7 SSE7.1 demo代码7.2 生产环境的应用 #xff08;重要#xff09; 8 MQTT 1 问题背景 扩宽自己的知识广度#xff0c;研究一下web实时消息推送 2 前言 文章参考自Web 实时消… 文章目录 1 问题背景2 前言3 什么是消息推送4 短轮询5 长轮询5.1 demo代码 6 iframe流6.1 demo代码 7 SSE7.1 demo代码7.2 生产环境的应用 重要 8 MQTT 1 问题背景 扩宽自己的知识广度研究一下web实时消息推送 2 前言 文章参考自Web 实时消息推送的 7 种实现方案针对一些比较重要的方式我都会尽量敲出一份完整的demo代码享受其中的编程乐趣。在SSE方式中笔者延申思考将他应用于电商支付的场景中给出了比较合理的解决方案但并未在生产环境中验证仍待考证。 3 什么是消息推送
消息推送是指服务端将消息推送给客户端。常见的场景有有人关注公众号公众号推送消息给关注者站内消息通知未读邮件数量监控告警数量等等。
4 短轮询 常见的http请求即是短轮询由客户端发起请求服务端接收请求并同步实时处理最后返回数据给客户端。 5 长轮询 短轮询的异步方式即是长轮询异步在哪里客户端发起请求web容器比如tomcat安排子线程去处理这些请求将这些请求交给服务端后无需阻塞等待结果tomcat会立即安排该子线程理其他请求 tomcat以此接收更多的请求提升系统的吞吐量。服务端处理完请求再返回数据给客户端。 5.1 demo代码 因为一个ID可能会被多个长轮询请求监听所以采用了guava包提供的Multimap结构存放长轮询一个key可以对应多个value。一旦监听到key发生变化对应的所有长轮询都会响应。 引入guava依赖 dependencygroupIdcom.google.guava/groupIdartifactIdguava/artifactIdversion31.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 MultimapString, DeferredResultString watchRequests Multimaps.synchronizedMultimap(HashMultimap.create());public static final long TIME_OUT 100000;/*** 设置监听*/GetMapping(path watch/{id})ResponseBodypublic DeferredResultString watch(PathVariable String id) {// 延迟对象设置超时时间DeferredResultString 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)) {CollectionDeferredResultString deferredResults watchRequests.get(id);for (DeferredResultString 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)ResponseBodyExceptionHandler(AsyncRequestTimeoutException.class)public String asyncRequestTimeoutHandler(AsyncRequestTimeoutException e) {System.out.println(异步请求超时);return 304;}
} 测试 首先页面发起长轮询请求/polling/watch/10086监听消息更变请求被挂起不变更数据直至超时再次发起了长轮询请求紧接着手动变更数据/polling/publish/10086长轮询得到响应前端处理业务逻辑完成后再次发起请求如此循环往复。
6 iframe流 在页面中插入一个隐藏的iframe标签通过在src中请求消息数量API接口由此在服务端和客户端之间创建一条长连接服务端持续向iframe传输数据。传输的数据通常是htmljs脚本。 6.1 demo代码 笔者打算使用一个页面来展示效果因此需要引入一个freemarker依赖 dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-freemarker/artifactId
/dependencyyml中配置freemarker spring:freemarker:suffix: .ftlcontent-type: text/htmlcharset: UTF-8cache: false# ftl页面存放的路径template-loader-path: classpath:/templates/写一个ftl页面 ftl页面源码如下 !DOCTYPE html
html langen
headmeta charsetUTF-8titleIFRAME/title
/head
body
iframe src/iframe/message styledisplay:none/iframe
divh1clock/h1span idclock/spanh1count/h1span idcount/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有如下几个特点 基于HTTP单向通信实现简单无需引入其组件默认支持断线重连 服务端重启客户端会重新发送连接请求这是天生解决服务端发版的问题啊只能传送文本信息
7.1 demo代码 SSE同样是使用页面展示效果需要添加一些freemarker相关东西具体细节可见第6.1节 页面代码 页面源码 !DOCTYPE html
html langen
headmeta charsetUTF-8titleIFRAME/title
/head
body
div idmessage/div
scriptlet source null;// 获取url中userId参数的值。如urlhttp://localhost:8033/sse/index?userId1111var 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 你的浏览器不支持SSEbr;}
/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 MapString, 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?userId1111注册客户端连接此处记为index页面。
浏览器另开一个tab页访问http://localhost:8033/sse/message/1111?msghaha然后去index页面看会发现有消息展示出来了 当服务端重启客户端会自动重连即index页面的那个http请求会再次发给服务端
7.2 生产环境的应用 重要 笔者做支付比较多该sse方式也可以用于做支付结果的消息通知一般都是用短轮询做查询查询支付结果那么现在可以使用SSE方式。针对应用于生产环境笔者认为有如下几点需要注意 由前面服务端代码可见服务端需要在内存中保存客户端的连接那个sseEmitterMap。在服务端是集群的情况下接收客户端请求的服务端节点的内存中并不一定就有客户端的连接此处可以使用Redis的发布订阅功能通知存有客户端连接的服务端节点进行发消息。除了Redis发布订阅还能通过RedisRPC做一个精准调用Redis可以存储Map客户端连接的唯一标识, 服务端节点IP拿到IP后通过RPC进行精准调用详情可以见服务端实时推送技术之SSEServer-Send Events。 服务端节点中的每个客户端连接都需要做超时处理超时则将连接从内存中移除否则会发生OOM。 假如服务端发版内存中的所有客户端连接都会丢失但无需担忧因为客户端默认会重连。
8 MQTT MQTT方式需要借助消息队列来实现其实相当于常规的生产者消费者模式。因实现起来比较复杂需要搭建MQ笔者此处暂不研究MQTT具体实现。
