从构建分布式秒杀系统聊聊WebSocket推送通知
前言
秒杀架构到后期,我们采用了消息队列的形式实现抢购逻辑,那么之前抛出过这样一个问题:消息队列异步处理完每个用户请求后,如何通知给相应用户秒杀成功?
场景映射
首先,我们举一个生活中比较常见的例子:我们去银行办理业务,一般会选择相关业务打印一个排号纸,然后就可以坐在小板凳上玩着手机,等待被小喇叭报号。当小喇叭喊到你所持有的号码,就可以拿着排号纸去柜台办理自己的业务。
这里,假设当我们取排号纸的时候,银行根据时间段内的排队情况,比较人性化的提示用户:排队人数较多,您是否继续等待?否的话我们可以换个时间段再来办理。
由此我们把生活场景映射到真实的秒杀业务逻辑中来:
- 我们可以把柜台比喻成商品下单处理逻辑单元
- 拿到排号纸说明你进入相应商品处理队列
- 拿到排号纸的请求直接返回前台,提示用户抢购进行中
- 排号纸进入队列后,等待商品业务处理逻辑
- 小喇叭叫到自己的排号相当于服务端通知用户秒杀成功,这时候可以进行支付逻辑
- 那些拿不到票号的同学,相当于队列已满直接返回秒杀失败
解决方案
通过上面的场景,我们很容易能够想到一种方案就是服务端通知,那么如何做到服务端异步通知的呢?下面,主角开始登场了,就是我们的Websocket。
WebSocket是HTML5开始提供的一种浏览器与服务器间进行全双工通讯的网络技术。依靠这种技术可以实现客户端和服务器端的长连接,双向实时通信。
特点:
- 异步、事件触发
- 可以发送文本,图片等流文件
- 数据格式比较轻量,性能开销小,通信高效
- 使用ws或者wss协议的客户端socket,能够实现真正意义上的推送功能
缺点:
- 部分浏览器不支持,浏览器支持的程度与方式有区别,需要各种兼容写法。
集成案例
由于我们的秒杀架构项目案例中使用了SpringBoot,因此集成webSocket也是相对比较简单的。
首先pom.xml引入以下依赖:
<!-- webSocket 秒杀通知--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-websocket</artifactId> </dependency>
WebSocketConfig 配置:
/** * WebSocket配置 * 创建者 爪哇笔记 * 创建时间 2018年5月29日 */ @Configuration public class WebSocketConfig { @Bean public ServerEndpointExporter serverEndpointExporter() { return new ServerEndpointExporter(); } }
WebSocketServer 配置:
@ServerEndpoint("/websocket/{userId}") @Component public class WebSocketServer { private final static Logger log = LoggerFactory.getLogger(WebSocketServer.class); //静态变量,用来记录当前在线连接数。应该把它设计成线程安全的。 private static int onlineCount = 0; //concurrent包的线程安全Set,用来存放每个客户端对应的MyWebSocket对象。 private static CopyOnWriteArraySet<WebSocketServer> webSocketSet = new CopyOnWriteArraySet<WebSocketServer>(); //与某个客户端的连接会话,需要通过它来给客户端发送数据 private Session session; //接收userId private String userId=""; /** * 连接建立成功调用的方法*/ @OnOpen public void onOpen(Session session,@PathParam("userId") String userId) { this.session = session; webSocketSet.add(this); //加入set中 addOnlineCount(); //在线数加1 log.info("有新窗口开始监听:"+userId+",当前在线人数为" + getOnlineCount()); this.userId=userId; try { sendMessage("连接成功"); } catch (IOException e) { log.error("websocket IO异常"); } } /** * 连接关闭调用的方法 */ @OnClose public void onClose() { webSocketSet.remove(this); //从set中删除 subOnlineCount(); //在线数减1 log.info("有一连接关闭!当前在线人数为" + getOnlineCount()); } /** * 收到客户端消息后调用的方法 * @param message 客户端发送过来的消息*/ @OnMessage public void onMessage(String message, Session session) { log.info("收到来自窗口"+userId+"的信息:"+message); //群发消息 for (WebSocketServer item : webSocketSet) { try { item.sendMessage(message); } catch (IOException e) { e.printStackTrace(); } } } /** * @param session * @param error */ @OnError public void onError(Session session, Throwable error) { log.error("发生错误"); error.printStackTrace(); } /** * 实现服务器主动推送 */ public void sendMessage(String message) throws IOException { this.session.getBasicRemote().sendText(message); } /** * 群发自定义消息 * */ public static void sendInfo(String message,@PathParam("userId") String userId){ log.info("推送消息到窗口"+userId+",推送内容:"+message); for (WebSocketServer item : webSocketSet) { try { //这里可以设定只推送给这个userId的,为null则全部推送 if(userId==null) { item.sendMessage(message); }else if(item.userId.equals(userId)){ item.sendMessage(message); } } catch (IOException e) { continue; } } } public static synchronized int getOnlineCount() { return onlineCount; } public static synchronized void addOnlineCount() { WebSocketServer.onlineCount++; } public static synchronized void subOnlineCount() { WebSocketServer.onlineCount--; } }
KafkaConsumer 消费配置,通知用户是否秒杀成功:
/** * 消费者 spring-kafka 2.0 + 依赖JDK8 * @author 科帮网 By https://blog.52itstyle.com */ @Component public class KafkaConsumer { @Autowired private ISeckillService seckillService; private static RedisUtil redisUtil = new RedisUtil(); /** * 监听seckill主题,有消息就读取 * @param message */ @KafkaListener(topics = {"seckill"}) public void receiveMessage(String message){ //收到通道的消息之后执行秒杀操作 String[] array = message.split(";"); if(redisUtil.getValue(array[0])!=null){//control层已经判断了,其实这里不需要再判断了 Result result = seckillService.startSeckil(Long.parseLong(array[0]), Long.parseLong(array[1])); if(result.equals(Result.ok())){ WebSocketServer.sendInfo(array[0].toString(), "秒杀成功");//推送给前台 }else{ WebSocketServer.sendInfo(array[0].toString(), "秒杀失败");//推送给前台 redisUtil.cacheValue(array[0], "ok");//秒杀结束 } }else{ WebSocketServer.sendInfo(array[0].toString(), "秒杀失败");//推送给前台 } } }
webSocket.js 前台通知逻辑:
$(function(){ socket.init(); }); var basePath = "ws://localhost:8080/seckill/"; socket = { webSocket : "", init : function() { //userId:自行追加 if ('WebSocket' in window) { webSocket = new WebSocket(basePath+'websocket/1'); } else if ('MozWebSocket' in window) { webSocket = new MozWebSocket(basePath+"websocket/1"); } else { webSocket = new SockJS(basePath+"sockjs/websocket"); } webSocket.onerror = function(event) { alert("websockt连接发生错误,请刷新页面重试!") }; webSocket.onopen = function(event) { }; webSocket.onmessage = function(event) { var message = event.data; alert(message)//判断秒杀是否成功、自行处理逻辑 }; } }
客户端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
低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。
持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。
- 上一篇
使用Python boto3上传Windows EC2实例中的文件至S3存储桶中
一、创建终端节点 为什么要创建终端节点,把VPC和S3管理起来呢?如果不将VPC和S3通过终端节点管理起来,那么VPC中EC2实例访问S3存储桶是通过公共网络的;一旦关联起来,那么VPC中EC2实例访问S3存储桶走的就是内部网络。好处有两个:1. 走内部网络就不会产生流量费用;2. 走内部网络速度快,不会因为网络原因导致我们的Python脚本产生异常。 VPC->终端节点->创建终端节点->将VPC和S3关联->关联子网 二、在Windows中安装Python3编译器以及boto3库 1. 下载地址:https://www.python.org/ 2. 双击安装,默认安装路径“C:\Users\用户\AppData\Local\Programs\Python\Python36” 3. 配置环境变量 4. 安装boto3开发库(环境变量配好即可使用pip命令) 三、生成AWS IAM用户密钥并配置 1. IAM->用户->选择具有访问S3权限的用户->安全证书->创建访问安全密钥->下载密钥文件到本地 2. 在Windows实例...
- 下一篇
Mysql主从架构-主库宕机如何恢复业务
在我们日常工作场景,首先要做到架构无单点隐患,其次在优化【安全、性能、高可用、高并发等】,Mysql这款关系型数据库稳定、高效,所以使用广泛,如果企业架构是1主多从,那如果Mysql主库宕机,如何解决? 最新博客地址: https://www.dgstack.cn/archives/178.html ----MySQL 主从同步原理图 一、Mysql主库宕机情况分类: 1)硬件问题,(服务器、ecs、虚拟主机等等)宕机 2)service问题,Mysql宕机,服务异常,端口异常等 二、硬件问题处理思路 硬件问题我们可以查看IDC巡检记录,或通过远程控制卡查看硬件运行状态,根据事实情况就行硬件故障报修进行处理,恢复业务步骤: 1)查看报警信息,确认业务是否收到影响,必要时切从库进行数据交互 2)IDC询问排查 3)确认硬件故障,短时间无法修复开Case处理 4)通知部门领导,处理进度,并实时记录 5)事件处理完成后,拟写故障报告,会议通报。 三、MySQL service问题处理思路 1)首先要做的就是判断是否影响业务,是否需要切库,保证业务运行时首要任务2)如果此时需要切从库,安装如下...
相关文章
文章评论
共有0条评论来说两句吧...
文章二维码
点击排行
推荐阅读
最新文章
- CentOS8编译安装MySQL8.0.19
- CentOS8,CentOS7,CentOS6编译安装Redis5.0.7
- SpringBoot2整合MyBatis,连接MySql数据库做增删改查操作
- SpringBoot2整合Redis,开启缓存,提高访问速度
- SpringBoot2配置默认Tomcat设置,开启更多高级功能
- Hadoop3单机部署,实现最简伪集群
- CentOS7,CentOS8安装Elasticsearch6.8.6
- CentOS6,7,8上安装Nginx,支持https2.0的开启
- Docker使用Oracle官方镜像安装(12C,18C,19C)
- SpringBoot2编写第一个Controller,响应你的http请求并返回结果