前言
秒杀架构到后期,我们采用了消息队列的形式实现抢购逻辑,那么之前抛出过这样一个问题:消息队列异步处理完每个用户请求后,如何通知给相应用户秒杀成功?
场景映射
首先,我们举一个生活中比较常见的例子:我们去银行办理业务,一般会选择相关业务打印一个排号纸,然后就可以坐在小板凳上玩着手机,等待被小喇叭报号。当小喇叭喊到你所持有的号码,就可以拿着排号纸去柜台办理自己的业务。
这里,假设当我们取排号纸的时候,银行根据时间段内的排队情况,比较人性化的提示用户:排队人数较多,您是否继续等待?否的话我们可以换个时间段再来办理。
由此我们把生活场景映射到真实的秒杀业务逻辑中来:
- 我们可以把柜台比喻成商品下单处理逻辑单元
- 拿到排号纸说明你进入相应商品处理队列
- 拿到排号纸的请求直接返回前台,提示用户抢购进行中
- 排号纸进入队列后,等待商品业务处理逻辑
- 小喇叭叫到自己的排号相当于服务端通知用户秒杀成功,这时候可以进行支付逻辑
- 那些拿不到票号的同学,相当于队列已满直接返回秒杀失败
解决方案
通过上面的场景,我们很容易能够想到一种方案就是服务端通知,那么如何做到服务端异步通知的呢?下面,主角开始登场了,就是我们的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 (WebSocketSer