从构建分布式秒杀系统聊聊WebSocket推送通知
237 次阅读 - 文章,基础架构,编程语言前言
秒杀架构到后期，我们采用了消息队列的形式实现抢购逻辑，那么之前抛出过这样一个问题：消息队列异步处理完每个用户请求后，如何通知给相应用户秒杀成功？
场景映射
首先，我们举一个生活中比较常见的例子：我们去银行办理业务，一般会选择相关业务打印一个排号纸，然后就可以坐在小板凳上玩着手机，等待被小喇叭报号。当小喇叭喊到你所持有的号码，就可以拿着排号纸去柜台办理自己的业务。
这里，假设当我们取排号纸的时候，银行根据时间段内的排队情况，比较人性化的提示用户：排队人数较多，您是否继续等待？否的话我们可以换个时间段再来办理。
由此我们把生活场景映射到真实的秒杀业务逻辑中来：
我们可以把柜台比喻成商品下单处理逻辑单元
拿到排号纸说明你进入相应商品处理队列
拿到排号纸的请求直接返回前台，提示用户抢购进行中
排号纸进入队列后，等待商品业务处理逻辑
小喇叭叫到自己的排号相当于服务端通知用户秒杀成功，这时候可以进行支付逻辑
那些拿不到票号的同学，相当于队列已满直接返回秒杀失败
解决方案
通过上面的场景，我们很容易能够想到一种方案就是服务端通知，那么如何做到服务端异步通知的呢？下面，主角开始登场了，就是我们的Websocket。
WebSocket是HTML5开始提供的一种浏览器与服务器间进行全双工通讯的网络技术。依靠这种技术可以实现客户端和服务器端的长连接，双向实时通信。
特点:
异步、事件触发
可以发送文本，图片等流文件
数据格式比较轻量，性能开销小，通信高效
使用ws或者wss协议的客户端socket，能够实现真正意义上的推送功能
缺点：
部分浏览器不支持，浏览器支持的程度与方式有区别，需要各种兼容写法。
集成案例
由于我们的秒杀架构项目案例中使用了SpringBoot，因此集成webSocket也是相对比较简单的。
首先pom.xml引入以下依赖：
|
1 2 3 4 5 |
<span class="hljs-comment"><!-- webSocket 秒杀通知--></span> <span class="hljs-tag"><<span class="hljs-name">dependency</span>></span> <span class="hljs-tag"><<span class="hljs-name">groupId</span>></span>org.springframework.boot<span class="hljs-tag"></<span class="hljs-name">groupId</span>></span> <span class="hljs-tag"><<span class="hljs-name">artifactId</span>></span>spring-boot-starter-websocket<span class="hljs-tag"></<span class="hljs-name">artifactId</span>></span> <span class="hljs-tag"></<span class="hljs-name">dependency</span>></span> |
WebSocketConfig 配置：
|
1 2 3 4 5 6 7 8 9 10 11 12 |
<span class="hljs-comment">/** * WebSocket配置 * 创建者 爪哇笔记 * 创建时间 2018年5月29日 */</span> <span class="hljs-meta">@Configuration</span> <span class="hljs-keyword">public</span> <span class="hljs-class"><span class="hljs-keyword">class</span> <span class="hljs-title">WebSocketConfig</span> </span>{ <span class="hljs-meta">@Bean</span> <span class="hljs-function"><span class="hljs-keyword">public</span> ServerEndpointExporter <span class="hljs-title">serverEndpointExporter</span><span class="hljs-params">()</span> </span>{ <span class="hljs-keyword">return</span> <span class="hljs-keyword">new</span> ServerEndpointExporter(); } } |
WebSocketServer 配置：
|
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 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 |
<span class="hljs-meta">@ServerEndpoint</span>(<span class="hljs-string">"/websocket/{userId}"</span>) <span class="hljs-meta">@Component</span> <span class="hljs-keyword">public</span> <span class="hljs-class"><span class="hljs-keyword">class</span> <span class="hljs-title">WebSocketServer</span> </span>{ <span class="hljs-keyword">private</span> <span class="hljs-keyword">final</span> <span class="hljs-keyword">static</span> Logger log = LoggerFactory.getLogger(WebSocketServer.class); <span class="hljs-comment">//静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。</span> <span class="hljs-keyword">private</span> <span class="hljs-keyword">static</span> <span class="hljs-keyword">int</span> onlineCount = <span class="hljs-number">0</span>; <span class="hljs-comment">//concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。</span> <span class="hljs-keyword">private</span> <span class="hljs-keyword">static</span> CopyOnWriteArraySet<WebSocketServer> webSocketSet = <span class="hljs-keyword">new</span> CopyOnWriteArraySet<WebSocketServer>(); <span class="hljs-comment">//与某个客户端的连接会话，需要通过它来给客户端发送数据</span> <span class="hljs-keyword">private</span> Session session; <span class="hljs-comment">//接收userId</span> <span class="hljs-keyword">private</span> String userId=<span class="hljs-string">""</span>; <span class="hljs-comment">/** * 连接建立成功调用的方法*/</span> <span class="hljs-meta">@OnOpen</span> <span class="hljs-function"><span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">onOpen</span><span class="hljs-params">(Session session,@PathParam(<span class="hljs-string">"userId"</span>)</span> String userId) </span>{ <span class="hljs-keyword">this</span>.session = session; webSocketSet.add(<span class="hljs-keyword">this</span>); <span class="hljs-comment">//加入set中</span> addOnlineCount(); <span class="hljs-comment">//在线数加1</span> log.info(<span class="hljs-string">"有新窗口开始监听:"</span>+userId+<span class="hljs-string">",当前在线人数为"</span> + getOnlineCount()); <span class="hljs-keyword">this</span>.userId=userId; <span class="hljs-keyword">try</span> { sendMessage(<span class="hljs-string">"连接成功"</span>); } <span class="hljs-keyword">catch</span> (IOException e) { log.error(<span class="hljs-string">"websocket IO异常"</span>); } } <span class="hljs-comment">/** * 连接关闭调用的方法 */</span> <span class="hljs-meta">@OnClose</span> <span class="hljs-function"><span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">onClose</span><span class="hljs-params">()</span> </span>{ webSocketSet.remove(<span class="hljs-keyword">this</span>); <span class="hljs-comment">//从set中删除</span> subOnlineCount(); <span class="hljs-comment">//在线数减1</span> log.info(<span class="hljs-string">"有一连接关闭！当前在线人数为"</span> + getOnlineCount()); } <span class="hljs-comment">/** * 收到客户端消息后调用的方法 * <span class="hljs-doctag">@param</span> message 客户端发送过来的消息*/</span> <span class="hljs-meta">@OnMessage</span> <span class="hljs-function"><span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">onMessage</span><span class="hljs-params">(String message, Session session)</span> </span>{ log.info(<span class="hljs-string">"收到来自窗口"</span>+userId+<span class="hljs-string">"的信息:"</span>+message); <span class="hljs-comment">//群发消息</span> <span class="hljs-keyword">for</span> (WebSocketServer item : webSocketSet) { <span class="hljs-keyword">try</span> { item.sendMessage(message); } <span class="hljs-keyword">catch</span> (IOException e) { e.printStackTrace(); } } } <span class="hljs-comment">/** * <span class="hljs-doctag">@param</span> session * <span class="hljs-doctag">@param</span> error */</span> <span class="hljs-meta">@OnError</span> <span class="hljs-function"><span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">onError</span><span class="hljs-params">(Session session, Throwable error)</span> </span>{ log.error(<span class="hljs-string">"发生错误"</span>); error.printStackTrace(); } <span class="hljs-comment">/** * 实现服务器主动推送 */</span> <span class="hljs-function"><span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">sendMessage</span><span class="hljs-params">(String message)</span> <span class="hljs-keyword">throws</span> IOException </span>{ <span class="hljs-keyword">this</span>.session.getBasicRemote().sendText(message); } <span class="hljs-comment">/** * 群发自定义消息 * */</span> <span class="hljs-function"><span class="hljs-keyword">public</span> <span class="hljs-keyword">static</span> <span class="hljs-keyword">void</span> <span class="hljs-title">sendInfo</span><span class="hljs-params">(String message,@PathParam(<span class="hljs-string">"userId"</span>)</span> String userId)</span>{ log.info(<span class="hljs-string">"推送消息到窗口"</span>+userId+<span class="hljs-string">"，推送内容:"</span>+message); <span class="hljs-keyword">for</span> (WebSocketServer item : webSocketSet) { <span class="hljs-keyword">try</span> { <span class="hljs-comment">//这里可以设定只推送给这个userId的，为null则全部推送</span> <span class="hljs-keyword">if</span>(userId==<span class="hljs-keyword">null</span>) { item.sendMessage(message); }<span class="hljs-keyword">else</span> <span class="hljs-keyword">if</span>(item.userId.equals(userId)){ item.sendMessage(message); } } <span class="hljs-keyword">catch</span> (IOException e) { <span class="hljs-keyword">continue</span>; } } } <span class="hljs-function"><span class="hljs-keyword">public</span> <span class="hljs-keyword">static</span> <span class="hljs-keyword">synchronized</span> <span class="hljs-keyword">int</span> <span class="hljs-title">getOnlineCount</span><span class="hljs-params">()</span> </span>{ <span class="hljs-keyword">return</span> onlineCount; } <span class="hljs-function"><span class="hljs-keyword">public</span> <span class="hljs-keyword">static</span> <span class="hljs-keyword">synchronized</span> <span class="hljs-keyword">void</span> <span class="hljs-title">addOnlineCount</span><span class="hljs-params">()</span> </span>{ WebSocketServer.onlineCount++; } <span class="hljs-function"><span class="hljs-keyword">public</span> <span class="hljs-keyword">static</span> <span class="hljs-keyword">synchronized</span> <span class="hljs-keyword">void</span> <span class="hljs-title">subOnlineCount</span><span class="hljs-params">()</span> </span>{ WebSocketServer.onlineCount--; } } |
KafkaConsumer 消费配置，通知用户是否秒杀成功：
|
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 |
<span class="hljs-comment">/** * 消费者 spring-kafka 2.0 + 依赖JDK8 * <span class="hljs-doctag">@author</span> 科帮网 By https://blog.52itstyle.com */</span> <span class="hljs-meta">@Component</span> <span class="hljs-keyword">public</span> <span class="hljs-class"><span class="hljs-keyword">class</span> <span class="hljs-title">KafkaConsumer</span> </span>{ <span class="hljs-meta">@Autowired</span> <span class="hljs-keyword">private</span> ISeckillService seckillService; <span class="hljs-keyword">private</span> <span class="hljs-keyword">static</span> RedisUtil redisUtil = <span class="hljs-keyword">new</span> RedisUtil(); <span class="hljs-comment">/** * 监听seckill主题,有消息就读取 * <span class="hljs-doctag">@param</span> message */</span> <span class="hljs-meta">@KafkaListener</span>(topics = {<span class="hljs-string">"seckill"</span>}) <span class="hljs-function"><span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">receiveMessage</span><span class="hljs-params">(String message)</span></span>{ <span class="hljs-comment">//收到通道的消息之后执行秒杀操作</span> String[] array = message.split(<span class="hljs-string">";"</span>); <span class="hljs-keyword">if</span>(redisUtil.getValue(array[<span class="hljs-number">0</span>])!=<span class="hljs-keyword">null</span>){<span class="hljs-comment">//control层已经判断了，其实这里不需要再判断了</span> Result result = seckillService.startSeckil(Long.parseLong(array[<span class="hljs-number">0</span>]), Long.parseLong(array[<span class="hljs-number">1</span>])); <span class="hljs-keyword">if</span>(result.equals(Result.ok())){ WebSocketServer.sendInfo(array[<span class="hljs-number">0</span>].toString(), <span class="hljs-string">"秒杀成功"</span>);<span class="hljs-comment">//推送给前台</span> }<span class="hljs-keyword">else</span>{ WebSocketServer.sendInfo(array[<span class="hljs-number">0</span>].toString(), <span class="hljs-string">"秒杀失败"</span>);<span class="hljs-comment">//推送给前台</span> redisUtil.cacheValue(array[<span class="hljs-number">0</span>], <span class="hljs-string">"ok"</span>);<span class="hljs-comment">//秒杀结束</span> } }<span class="hljs-keyword">else</span>{ WebSocketServer.sendInfo(array[<span class="hljs-number">0</span>].toString(), <span class="hljs-string">"秒杀失败"</span>);<span class="hljs-comment">//推送给前台</span> } } } |
webSocket.js 前台通知逻辑：
|
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 |
$(<span class="hljs-function"><span class="hljs-keyword">function</span>()</span>{ socket.init(); }); <span class="hljs-keyword">var</span> basePath = <span class="hljs-string">"ws://localhost:8080/seckill/"</span>; socket = { <span class="hljs-attr">webSocket</span> : <span class="hljs-string">""</span>, <span class="hljs-attr">init</span> : <span class="hljs-function"><span class="hljs-keyword">function</span>() </span>{ <span class="hljs-comment">//userId：自行追加</span> <span class="hljs-keyword">if</span> (<span class="hljs-string">'WebSocket'</span> <span class="hljs-keyword">in</span> <span class="hljs-built_in">window</span>) { webSocket = <span class="hljs-keyword">new</span> WebSocket(basePath+<span class="hljs-string">'websocket/1'</span>); } <span class="hljs-keyword">else</span> <span class="hljs-keyword">if</span> (<span class="hljs-string">'MozWebSocket'</span> <span class="hljs-keyword">in</span> <span class="hljs-built_in">window</span>) { webSocket = <span class="hljs-keyword">new</span> MozWebSocket(basePath+<span class="hljs-string">"websocket/1"</span>); } <span class="hljs-keyword">else</span> { webSocket = <span class="hljs-keyword">new</span> SockJS(basePath+<span class="hljs-string">"sockjs/websocket"</span>); } webSocket.onerror = <span class="hljs-function"><span class="hljs-keyword">function</span>(<span class="hljs-params">event</span>) </span>{ alert(<span class="hljs-string">"websockt连接发生错误，请刷新页面重试!"</span>) }; webSocket.onopen = <span class="hljs-function"><span class="hljs-keyword">function</span>(<span class="hljs-params">event</span>) </span>{ }; webSocket.onmessage = <span class="hljs-function"><span class="hljs-keyword">function</span>(<span class="hljs-params">event</span>) </span>{ <span class="hljs-keyword">var</span> message = event.data; alert(message)<span class="hljs-comment">//判断秒杀是否成功、自行处理逻辑</span> }; } } |
客户端API
客户端与服务器通信
send() 向远程服务器发送数据
close() 关闭该websocket链接
监听函数
onopen 当网络连接建立时触发该事件
onerror 当网络发生错误时触发该事件
onclose 当websocket被关闭时触发该事件
onmessage 当websocket接收到服务器发来的消息的时触发的事件，也是通信中最重要的一个监听事件。msg.data
readyState属性
这个属性可以返回websocket所处的状态。
CONNECTING(0) websocket正尝试与服务器建立连接
OPEN(1) websocket与服务器已经建立连接
CLOSING(2) websocket正在关闭与服务器的连接
CLOSED(3) websocket已经关闭了与服务器的连接
开源方案
goeasy
GoEasy实时Web推送，支持后台推送和前台推送两种：后台推送可以选择Java SDK、 Restful API支持所有开发语言；前台推送：JS推送。无论选择哪种方式推送代码都十分简单（10分钟可搞定）。由于它支持websocket 和polling两种连接方式所以兼顾大多数主流浏览器，低版本的IE浏览器也是支持的。
Pushlets
Pushlets 是通过长连接方式实现“推”消息的。推送模式分为：Poll（轮询）、Pull（拉）。
Pushlet
Pushlet 是一个开源的 Comet 框架,Pushlet 使用了观察者模型：客户端发送请求，订阅感兴趣的事件；服务器端为每个客户端分配一个会话 ID 作为标记，事件源会把新产生的事件以多播的方式发送到订阅者的事件队列里。
地址：https://github.com/wjw465150/Pushlet
总结
其实前面有提过，尽管WebSocket有诸多优点，但是，如果服务端维护很多长连接也是挺耗费资源的，服务器集群以及览器或者客户端兼容性问题，也会带来了一些不确定性因素。大体了解了一下各大厂的做法，大多数都还是基于轮询的方式实现的，比如：腾讯PC端微信扫码登录、京东商城支付成功通知等等。
有些小伙伴可能会问了，轮询岂不是会更耗费资源？其实在我看来，有些轮询是不可能穿透到后端数据库查询服务的，比如秒杀，一个缓存标记位就可以判定是否秒杀成功。相对于WS的长连接以及其不确定因素，在秒杀场景下，轮询还是相对比较合适的。
思考
最后，思考一个问题：100件商品，假如有一万人进行抢购，该如何设置队列长度？
秒杀案例：https://gitee.com/52itstyle/spring-boot-seckill
参考
https://blog.52itstyle.com/archives/736/
https://www.xoriant.com/blog/mobility/websocket-web-stateful-now.html
文章来源：51CTO博客作者小柒2015
注：转载文章均来自于公开网络，仅供学习使用，不会用于任何商业用途，如果侵犯到原作者的权益，请您与我们联系删除或者授权事宜，联系邮箱：contact@dataunion.org。转载数盟网站文章请注明原文章作者，否则产生的任何版权纠纷与数盟无关。
相关文章！
期待你一针见血的评论，Come on！