从构建分布式秒杀系统聊聊WebSocket推送通知

前言

秒杀架构到后期,我们采用了消息队列的形式实现抢购逻辑,那么之前抛出过这样一个问题:消息队列异步处理完每个用户请求后,如何通知给相应用户秒杀成功?

场景映射

首先,我们举一个生活中比较常见的例子:我们去银行办理业务,一般会选择相关业务打印一个排号纸,然后就可以坐在小板凳上玩着手机,等待被小喇叭报号。当小喇叭喊到你所持有的号码,就可以拿着排号纸去柜台办理自己的业务。

这里,假设当我们取排号纸的时候,银行根据时间段内的排队情况,比较人性化的提示用户:排队人数较多,您是否继续等待?否的话我们可以换个时间段再来办理。

由此我们把生活场景映射到真实的秒杀业务逻辑中来:

我们可以把柜台比喻成商品下单处理逻辑单元

拿到排号纸说明你进入相应商品处理队列

拿到排号纸的请求直接返回前台,提示用户抢购进行中

排号纸进入队列后,等待商品业务处理逻辑

小喇叭叫到自己的排号相当于服务端通知用户秒杀成功,这时候可以进行支付逻辑

那些拿不到票号的同学,相当于队列已满直接返回秒杀失败

解决方案

通过上面的场景,我们很容易能够想到一种方案就是服务端通知,那么如何做到服务端异步通知的呢?下面,主角开始登场了,就是我们的Websocket。

WebSocket是HTML5开始提供的一种浏览器与服务器间进行全双工通讯的网络技术。依靠这种技术可以实现客户端和服务器端的长连接,双向实时通信。

特点:

异步、事件触发

可以发送文本,图片等流文件

数据格式比较轻量,性能开销小,通信高效

使用ws或者wss协议的客户端socket,能够实现真正意义上的推送功能

缺点:

部分浏览器不支持,浏览器支持的程度与方式有区别,需要各种兼容写法。

集成案例

由于我们的秒杀架构项目案例中使用了SpringBoot,因此集成webSocket也是相对比较简单的。

首先pom.xml引入以下依赖:

org.springframework.bootspring-boot-starter-websocket

WebSocketConfig 配置:

/**

* WebSocket配置

* 创建者  爪哇笔记

* 创建时间 2018年5月29日

*/@ConfigurationpublicclassWebSocketConfig{@BeanpublicServerEndpointExporterserverEndpointExporter(){returnnewServerEndpointExporter();      }  }

WebSocketServer 配置:

@ServerEndpoint("/websocket/{userId}")@ComponentpublicclassWebSocketServer{privatefinalstaticLogger log = LoggerFactory.getLogger(WebSocketServer.class);//静态变量,用来记录当前在线连接数。应该把它设计成线程安全的。privatestaticintonlineCount =0;//concurrent包的线程安全Set,用来存放每个客户端对应的MyWebSocket对象。privatestaticCopyOnWriteArraySet webSocketSet =newCopyOnWriteArraySet();//与某个客户端的连接会话,需要通过它来给客户端发送数据privateSession session;//接收userIdprivateString userId="";/**

    * 连接建立成功调用的方法*/@OnOpenpublicvoidonOpen(Session session,@PathParam("userId")String userId){this.session = session;        webSocketSet.add(this);//加入set中addOnlineCount();//在线数加1log.info("有新窗口开始监听:"+userId+",当前在线人数为"+ getOnlineCount());this.userId=userId;try{            sendMessage("连接成功");        }catch(IOException e) {            log.error("websocket IO异常");        }    }/**

    * 连接关闭调用的方法

    */@OnClosepublicvoidonClose(){        webSocketSet.remove(this);//从set中删除subOnlineCount();//在线数减1log.info("有一连接关闭!当前在线人数为"+ getOnlineCount());    }/**    * 收到客户端消息后调用的方法    *@parammessage 客户端发送过来的消息*/@OnMessagepublicvoidonMessage(String message, Session session){        log.info("收到来自窗口"+userId+"的信息:"+message);//群发消息for(WebSocketServer item : webSocketSet) {try{                item.sendMessage(message);            }catch(IOException e) {                e.printStackTrace();            }        }    }/**    *@paramsession    *@paramerror    */@OnErrorpublicvoidonError(Session session, Throwable error){        log.error("发生错误");        error.printStackTrace();    }/**

    * 实现服务器主动推送

    */publicvoidsendMessage(String message)throwsIOException{this.session.getBasicRemote().sendText(message);    }/**

    * 群发自定义消息

    * */publicstaticvoidsendInfo(String message,@PathParam("userId")String userId){        log.info("推送消息到窗口"+userId+",推送内容:"+message);for(WebSocketServer item : webSocketSet) {try{//这里可以设定只推送给这个userId的,为null则全部推送if(userId==null) {                    item.sendMessage(message);                }elseif(item.userId.equals(userId)){                    item.sendMessage(message);                }            }catch(IOException e) {continue;            }        }    }publicstaticsynchronizedintgetOnlineCount(){returnonlineCount;    }publicstaticsynchronizedvoidaddOnlineCount(){        WebSocketServer.onlineCount++;    }publicstaticsynchronizedvoidsubOnlineCount(){        WebSocketServer.onlineCount--;    }}

KafkaConsumer 消费配置,通知用户是否秒杀成功:

/** * 消费者 spring-kafka 2.0 + 依赖JDK8 *@author科帮网 By https://blog.52itstyle.com */@ComponentpublicclassKafkaConsumer{@AutowiredprivateISeckillService seckillService;privatestaticRedisUtil redisUtil =newRedisUtil();/**    * 监听seckill主题,有消息就读取    *@parammessage    */@KafkaListener(topics = {"seckill"})publicvoidreceiveMessage(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();});varbasePath ="ws://localhost:8080/seckill/";socket = {webSocket:"",init:function(){//userId:自行追加if('WebSocket'inwindow) {            webSocket =newWebSocket(basePath+'websocket/1');        }elseif('MozWebSocket'inwindow) {            webSocket =newMozWebSocket(basePath+"websocket/1");        }else{            webSocket =newSockJS(basePath+"sockjs/websocket");        }        webSocket.onerror =function(event){            alert("websockt连接发生错误,请刷新页面重试!")        };        webSocket.onopen =function(event){        };        webSocket.onmessage =function(event){varmessage = 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 作为标记,事件源会把新产生的事件以多播的方式发送到订阅者的事件队列里。


总结

其实前面有提过,尽管WebSocket有诸多优点,但是,如果服务端维护很多长连接也是挺耗费资源的,服务器集群以及览器或者客户端兼容性问题,也会带来了一些不确定性因素。大体了解了一下各大厂的做法,大多数都还是基于轮询的方式实现的,比如:腾讯PC端微信扫码登录、京东商城支付成功通知等等。

有些小伙伴可能会问了,轮询岂不是会更耗费资源?其实在我看来,有些轮询是不可能穿透到后端数据库查询服务的,比如秒杀,一个缓存标记位就可以判定是否秒杀成功。相对于WS的长连接以及其不确定因素,在秒杀场景下,轮询还是相对比较合适的。

1、具有1-5工作经验的,面对目前流行的技术不知从何下手,

需要突破技术瓶颈的可以加。

2、在公司待久了,过得很安逸,

但跳槽时面试碰壁。

需要在短时间内进修、跳槽拿高薪的可以加。

3、如果没有工作经验,但基础非常扎实,对java工作机制,

常用设计思想,常用java开发框架掌握熟练的,可以加。

4、觉得自己很牛B,一般需求都能搞定。

但是所学的知识点没有系统化,很难在技术领域继续突破的可以加。

5. 群号:高级架构群 Java进阶群:180705916.备注好信息!送架构视频。

6.阿里Java高级大牛直播讲解知识点,分享知识,

多年工作经验的梳理和总结,带着大家全面、

©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 204,732评论 6 478
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 87,496评论 2 381
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 151,264评论 0 338
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 54,807评论 1 277
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 63,806评论 5 368
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,675评论 1 281
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 38,029评论 3 399
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,683评论 0 258
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 41,704评论 1 299
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,666评论 2 321
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,773评论 1 332
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,413评论 4 321
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 39,016评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,978评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,204评论 1 260
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 45,083评论 2 350
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,503评论 2 343

推荐阅读更多精彩内容