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

首先,我们举一个生活中比较常见的例子:我们去银行办理业务,一般会选择相关业务打印一个排号纸,然后就可以坐在小板凳上玩着手机,等待被小喇叭报号。当小喇叭喊到你所持有的号码,就可以拿着排号纸去柜台办理自己的业务。
这里,假设当我们取排号纸的时候,银行根据时间段内的排队情况,比较人性化的提示用户:排队人数较多,您是否继续等待?否的话我们可以换个时间段再来办理。
由此我们把生活场景映射到真实的秒杀业务逻辑中来:
解决方案
通过上面的场景,我们很容易能够想到一种方案就是服务端通知,那么如何做到服务端异步通知的呢?下面,主角开始登场了,就是我们的Websocket。

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

HTTP VS WebSocket
特点:
缺点:
集成案例
由于我们的秒杀架构项目案例中使用了SpringBoot,因此集成webSocket也是相对比较简单的。
首先pom.xml引入以下依赖:
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
WebSocketConfig 配置:
@Configuration
publicclassWebSocketConfig{
@Bean
public ServerEndpointExporter serverEndpointExporter(){
returnnew ServerEndpointExporter();
}
}
WebSocketServer 配置:
@ServerEndpoint("/websocket/{userId}")
@Component
publicclassWebSocketServer{
privatefinalstatic Logger log = LoggerFactory.getLogger(WebSocketServer.class);
privatestaticint onlineCount = 0;
privatestatic CopyOnWriteArraySet<WebSocketServer> webSocketSet = new CopyOnWriteArraySet<WebSocketServer>();
private Session session;
private String userId="";
@OnOpen
publicvoidonOpen(Session session,@PathParam("userId") String userId) {
this.session = session;
webSocketSet.add(this);
addOnlineCount();
log.info("有新窗口开始监听:"+userId+",当前在线人数为" + getOnlineCount());
this.userId=userId;
try {
sendMessage("连接成功");
} catch (IOException e) {
log.error("websocket IO异常");
}
}
@OnClose
publicvoidonClose(){
webSocketSet.remove(this);
subOnlineCount();
log.info("有一连接关闭!当前在线人数为" + getOnlineCount());
}
@OnMessage
publicvoidonMessage(String message, Session session){
log.info("收到来自窗口"+userId+"的信息:"+message);
for (WebSocketServer item : webSocketSet) {
try {
item.sendMessage(message);
} catch (IOException e) {
e.printStackTrace();
}
}
}
@OnError
publicvoidonError(Session session, Throwable error){
log.error("发生错误");
error.printStackTrace();
}
publicvoidsendMessage(String message)throws IOException {
this.session.getBasicRemote().sendText(message); }
publicstaticvoidsendInfo(String message,@PathParam("userId") String userId){
log.info("推送消息到窗口"+userId+",推送内容:"+message);
for (WebSocketServer item : webSocketSet) {
try {
if(userId==null) {
item.sendMessage(message);
}elseif(item.userId.equals(userId)){
item.sendMessage(message);
}
} catch (IOException e) {
continue;
}
}
}
publicstaticsynchronizedintgetOnlineCount(){
return onlineCount;
}
publicstaticsynchronizedvoidaddOnlineCount(){
WebSocketServer.onlineCount++;
}
publicstaticsynchronizedvoidsubOnlineCount(){
WebSocketServer.onlineCount--;
}
}
KafkaConsumer 消费配置,通知用户是否秒杀成功:
@Component
publicclassKafkaConsumer{
@Autowired
private ISeckillService seckillService;
privatestatic RedisUtil redisUtil = new RedisUtil();
@KafkaListener(topics = {"seckill"})
publicvoidreceiveMessage(String message){
String[] array = message.split(";");
if(redisUtil.getValue(array[0])!=null){
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(){ soc
ket.init();});
var basePath = "ws://localhost:8080/seckill/";
socket = {
webSocket : "",
init : function() {
if ('WebSocket'inwindow) {
webSocket = new WebSocket(basePath+'websocket/1');
}
elseif ('MozWebSocket'inwindow) {
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链接
监听函数
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浏览器也是支持的。
地址:http://goeasy.io/
Pushlets
Pushlets 是通过长连接方式实现“推”消息的。推送模式分为:Poll(轮询)、Pull(拉)。
地址:http://www.pushlets.com/
Pushlet
Pushlet 是一个开源的 Comet 框架,Pushlet 使用了观察者模型:客户端发送请求,订阅感兴趣的事件;服务器端为每个客户端分配一个会话 ID 作为标记,事件源会把新产生的事件以多播的方式发送到订阅者的事件队列里。
地址:https://github.com/wjw465150/Pushlet
总结
其实前面有提过,尽管WebSocket有诸多优点,但是,如果服务端维护很多长连接也是挺耗费资源的,服务器集群以及览器或者客户端兼容性问题,也会带来了一些不确定性因素。大体了解了一下各大厂的做法,大多数都还是基于轮询的方式实现的,比如:腾讯PC端微信扫码登录、京东商城支付成功通知等等。
有些小伙伴可能会问了,轮询岂不是会更耗费资源?其实在我看来,有些轮询是不可能穿透到后端数据库查询服务的,比如秒杀,一个缓存标记位就可以判定是否秒杀成功。相对于WS的长连接以及其不确定因素,在秒杀场景下,轮询还是相对比较合适的。




