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

1,519 阅读9分钟
原文链接: mp.weixin.qq.com

前言

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

场景映射

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

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

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

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

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

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

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

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

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

解决方案

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

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

HTTP VS WebSocket

特点:

  • 异步、事件触发

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

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

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

缺点:

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

集成案例

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

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

  1. <!-- webSocket 秒杀通知-->

  2. <dependency>

  3.    <groupId>org.springframework.boot</groupId>

  4.    <artifactId>spring-boot-starter-websocket</artifactId>

  5. </dependency>

WebSocketConfig 配置:

  1. /**

  2. * WebSocket配置

  3. * 创建者  爪哇笔记

  4. * 创建时间    2018年5月29日

  5. */

  6. @Configuration  

  7. public class WebSocketConfig {  

  8.    @Bean  

  9.    public ServerEndpointExporter serverEndpointExporter() {  

  10.        return new ServerEndpointExporter();  

  11.    }  

  12. }

WebSocketServer 配置:

  1. @ServerEndpoint("/websocket/{userId}")

  2. @Component

  3. public class WebSocketServer {

  4.    private final static Logger log = LoggerFactory.getLogger(WebSocketServer.class);

  5.    //静态变量,用来记录当前在线连接数。应该把它设计成线程安全的。

  6.    private static int onlineCount = 0;

  7.    //concurrent包的线程安全Set,用来存放每个客户端对应的MyWebSocket对象。

  8.    private static CopyOnWriteArraySet<WebSocketServer> webSocketSet = new CopyOnWriteArraySet<WebSocketServer>();

  9.    //与某个客户端的连接会话,需要通过它来给客户端发送数据

  10.    private Session session;

  11.    //接收userId

  12.    private String userId="";

  13.    /**

  14.     * 连接建立成功调用的方法*/

  15.    @OnOpen

  16.    public void onOpen(Session session,@PathParam("userId") String userId) {

  17.        this.session = session;

  18.        webSocketSet.add(this);     //加入set中

  19.        addOnlineCount();           //在线数加1

  20.        log.info("有新窗口开始监听:"+userId+",当前在线人数为" + getOnlineCount());

  21.        this.userId=userId;

  22.        try {

  23.             sendMessage("连接成功");

  24.        } catch (IOException e) {

  25.            log.error("websocket IO异常");

  26.        }

  27.    }

  28.    /**

  29.     * 连接关闭调用的方法

  30.     */

  31.    @OnClose

  32.    public void onClose() {

  33.        webSocketSet.remove(this);  //从set中删除

  34.        subOnlineCount();           //在线数减1

  35.        log.info("有一连接关闭!当前在线人数为" + getOnlineCount());

  36.    }

  37.    /**

  38.     * 收到客户端消息后调用的方法

  39.     * @param message 客户端发送过来的消息*/

  40.    @OnMessage

  41.    public void onMessage(String message, Session session) {

  42.        log.info("收到来自窗口"+userId+"的信息:"+message);

  43.        //群发消息

  44.        for (WebSocketServer item : webSocketSet) {

  45.            try {

  46.                item.sendMessage(message);

  47.            } catch (IOException e) {

  48.                e.printStackTrace();

  49.            }

  50.        }

  51.    }

  52.    /**

  53.     * @param session

  54.     * @param error

  55.     */

  56.    @OnError

  57.    public void onError(Session session, Throwable error) {

  58.        log.error("发生错误");

  59.        error.printStackTrace();

  60.    }

  61.    /**

  62.     * 实现服务器主动推送

  63.     */

  64.    public void sendMessage(String message) throws IOException {

  65.        this.session.getBasicRemote().sendText(message);

  66.    }

  67.    /**

  68.     * 群发自定义消息

  69.     * */

  70.    public static void sendInfo(String message,@PathParam("userId") String userId){

  71.        log.info("推送消息到窗口"+userId+",推送内容:"+message);

  72.        for (WebSocketServer item : webSocketSet) {

  73.            try {

  74.                //这里可以设定只推送给这个userId的,为null则全部推送

  75.                if(userId==null) {

  76.                    item.sendMessage(message);

  77.                }else if(item.userId.equals(userId)){

  78.                    item.sendMessage(message);

  79.                }

  80.            } catch (IOException e) {

  81.                continue;

  82.            }

  83.        }

  84.    }

  85.    public static synchronized int getOnlineCount() {

  86.        return onlineCount;

  87.    }

  88.    public static synchronized void addOnlineCount() {

  89.        WebSocketServer.onlineCount++;

  90.    }

  91.    public static synchronized void subOnlineCount() {

  92.        WebSocketServer.onlineCount--;

  93.    }

  94. }

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

  1. /**

  2. * 消费者 spring-kafka 2.0 + 依赖JDK8

  3. * @author 科帮网 By https://blog.52itstyle.com

  4. */

  5. @Component

  6. public class KafkaConsumer {

  7.    @Autowired

  8.    private ISeckillService seckillService;

  9.    private static RedisUtil redisUtil = new RedisUtil();

  10.    /**

  11.     * 监听seckill主题,有消息就读取

  12.     * @param message

  13.     */

  14.    @KafkaListener(topics = {"seckill"})

  15.    public void receiveMessage(String message){

  16.        //收到通道的消息之后执行秒杀操作

  17.        String[] array = message.split(";");

  18.        if(redisUtil.getValue(array[0])!=null){//control层已经判断了,其实这里不需要再判断了

  19.            Result result = seckillService.startSeckil(Long.parseLong(array[0]), Long.parseLong(array[1]));

  20.            if(result.equals(Result.ok())){

  21.                WebSocketServer.sendInfo(array[0].toString(), "秒杀成功");//推送给前台

  22.            }else{

  23.                WebSocketServer.sendInfo(array[0].toString(), "秒杀失败");//推送给前台

  24.                redisUtil.cacheValue(array[0], "ok");//秒杀结束

  25.            }

  26.        }else{

  27.            WebSocketServer.sendInfo(array[0].toString(), "秒杀失败");//推送给前台

  28.        }

  29.    }

  30. }

webSocket.js 前台通知逻辑:

  1. $(function(){

  2.    socket.init();

  3. });

  4. var basePath = "ws://localhost:8080/seckill/";

  5. socket = {

  6.    webSocket : "",

  7.    init : function() {

  8.        //userId:自行追加

  9.        if ('WebSocket' in window) {

  10.            webSocket = new WebSocket(basePath+'websocket/1');

  11.        }

  12.        else if ('MozWebSocket' in window) {

  13.            webSocket = new MozWebSocket(basePath+"websocket/1");

  14.        }

  15.        else {

  16.            webSocket = new SockJS(basePath+"sockjs/websocket");

  17.        }

  18.        webSocket.onerror = function(event) {

  19.            alert("websockt连接发生错误,请刷新页面重试!")

  20.        };

  21.        webSocket.onopen = function(event) {

  22.        };

  23.        webSocket.onmessage = function(event) {

  24.            var message = event.data;

  25.            alert(message)//判断秒杀是否成功、自行处理逻辑

  26.        };

  27.    }

  28. }

客户端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浏览器也是支持的。

地址:http://goeasy.io/

Pushlets

Pushlets 是通过长连接方式实现“推”消息的。推送模式分为:Poll(轮询)、Pull(拉)。

地址:http://www.pushlets.com/

Pushlet

Pushlet 是一个开源的 Comet 框架,Pushlet 使用了观察者模型:客户端发送请求,订阅感兴趣的事件;服务器端为每个客户端分配一个会话 ID 作为标记,事件源会把新产生的事件以多播的方式发送到订阅者的事件队列里。

地址:https://github.com/wjw465150/Pushlet

总结

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

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

思考

最后思考一个问题:100件商品,假如有一万人进行抢购,该如何设置队列长度?

参考

https://blog.52itstyle.com/archives/736/

https://www.xoriant.com/blog/mobility/websocket-web-stateful-now.html