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

时间  2019-12-02
标签 构建 分布式 秒杀 系统 聊聊 websocket 推送 通知 栏目 系统架构 繁體版
原文   原文链接

摘要：
前言 秒杀架构到后期，咱们采用了消息队列的形式实现抢购逻辑，那么以前抛出过这样一个问题：消息队列异步处理完每一个用户请求后，如何通知给相应用户秒杀成功？ 场景映射 首先，咱们举一个生活中比较常见的例子：咱们去银行办理业务，通常会选择相关业务打印一个排号纸，而后就能够坐在小板凳上玩着手机，等待被小喇叭报号。

前言

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

场景映射

timg

首先，咱们举一个生活中比较常见的例子：咱们去银行办理业务，通常会选择相关业务打印一个排号纸，而后就能够坐在小板凳上玩着手机，等待被小喇叭报号。当小喇叭喊到你所持有的号码，就能够拿着排号纸去柜台办理本身的业务。git

这里，假设当咱们取排号纸的时候，银行根据时间段内的排队状况，比较人性化的提示用户：排队人数较多，您是否继续等待？否的话咱们能够换个时间段再来办理。github

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

  • 咱们能够把柜台比喻成商品下单处理逻辑单元
  • 拿到排号纸说明你进入相应商品处理队列
  • 拿到排号纸的请求直接返回前台，提示用户抢购进行中
  • 排号纸进入队列后，等待商品业务处理逻辑
  • 小喇叭叫到本身的排号至关于服务端通知用户秒杀成功，这时候能够进行支付逻辑
  • 那些拿不到票号的同窗，至关于队列已满直接返回秒杀失败

解决方案

经过上面的场景，咱们很容易可以想到一种方案就是服务端通知，那么如何作到服务端异步通知的呢？下面，主角开始登场了，就是咱们的Websocket。redis

1327868440

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

bg2017051502

特色:数据库

  • 异步、事件触发
  • 能够发送文本，图片等流文件
  • 数据格式比较轻量，性能开销小，通讯高效
  • 使用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浏览器也是支持的。

地址：goeasy.io/

Pushlets

Pushlets 是经过长链接方式实现“推”消息的。推送模式分为：Poll（轮询）、Pull（拉）。

地址：www.pushlets.com/

Pushlet

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

地址：github.com/wjw465150/P…

总结

其实前面有提过，尽管WebSocket有诸多优势，可是，若是服务端维护不少长链接也是挺耗费资源的，服务器集群以及览器或者客户端兼容性问题，也会带来了一些不肯定性因素。大致了解了一下各大厂的作法，大多数都仍是基于轮询的方式实现的，好比：腾讯PC端微信扫码登陆、京东商城支付成功通知等等。

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

思考

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

秒杀案例：gitee.com/52itstyle/s…

参考

blog.52itstyle.com/archives/73…

www.xoriant.com/blog/mobili…

原文连接

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程