问题：由于项目需要与web端进行通讯，web端需要获取同步信息的进度（由于同步过程可能会持续比较长时间，所以采用websocket进行通讯，不在需要web端定时轮询pull拉取进度信息）；
背景：
由于项目需要与web端进行通讯，web端需要获取同步信息的进度（由于同步过程可能会持续比较长时间，所以采用websocket进行通讯，不在需要web端定时轮询pull拉取进度信息）；
解决思路：
- 从整体上看是有两个思路：
前端通过定时器pull去服务端拉取同步信息，通过前端轮询会增加很多无效的请求，即是无效的查询，增加了很多无效的轮询；减少了工作量，加快了开发进度，但是进度同步给用户一个很不实时与直观的界面。
使用websocket，客户端与服务端增加socket通讯的，所以决定采用push的方式，与web端保持长连接，通过长连接来推送进度消息。 - 设备同步流程：
通过前端轮询会增加很多无效的请求，即是无效的查询，增加了很多无效的轮询；所以决定采用push的方式，与web端保持长连接，通过长连接来推送进度消息。
服务消息是通过web端发起http同步请求，建立socket连接，服务端进行异步处理需要同步的信息，在服务端进行同，通过Kafka发送进度消息，推送到对应消息中转服务，消息中转服务进行
分发到对应的push服务上（netty服务），netty服务上收到对应的消息，通过缓存对应用户的channl进行channel.writeAndFlush()把消息推送的到对应的长连接客户端上，完成整个流程。
方案介绍
对于长连接方案有很对种实现方式，当时也对于选型比较纠结，最终由于与web端交互，无法直接使用socket进行连接，使用了websocket进行推送。例如 IM，实时消息推送还有很多消息推送也使用了这个协议。
接下来我们比较一下websocket的框架，比较主流的有netty、tomcat、socketIO 三个框架。
基于支持websocket的容器，开发简单，例如tomcat，但在高并发的支持不是很好，连接的时候容易连接断开，还有就是依赖容器。
netty-socketIO是在netty4基础之上做了一层封装，效率如同netty一样，是一个全平台方案，友好的API，京东的logbook也是用了socketIO来传递日志，也是我们的一个备选方案。
netty是业内主流的NIO框架，netty对javaNIO做了封装，让开发者更多关注业务，降低开发成本，很多著名的RPC框架都采用了netty作为传输层，友好的API，功能强大，内置了很多编解码协议，实现websocket协议也是十分方便。
1、整个架构设计如下 image01.png
2. 推送服务会在业务服务端与netty之间建立一个分发机制，分发机制主要作用：
从redis中获取连接通道与netty服务的对应关系，就是对应通道推送到哪一台服务器上。
接收消息与分发消息（通过异步http分发或者kafka分发）。
3. 选型netty的原因，
- 由于对应社区的活跃度高；
- 服务相对稳定；
- 被大多数公司采纳，已经有一套完整的解决方案。
netty是具有异步非阻塞的特性，传统的IO是面向流的，NIO是面向缓冲区的，是非阻塞的。
netty使用的是reactor模型，有boss线程和worker线程。
- boss线程其实是一个独立的线程池，是专门负责接收client请求的，默认的线程池大小是1，由于我们只开一个端口，设置为默认值1就可以了。
- worker线程池通过名字就可以得知是工作线程池，专门负责处理读写操作，默认线程池的大小为2*cpu个数。
服务中的URL校验与权限校验Handler
1、校验请求URL
2、校验token
3、判断对应用户的连接数量
4、保存对应信息到缓存（1、channel的对象到本地缓存；2、用户对应的key，channel的list到redis）
public class HttpRequestHandler extends SimpleChannelInboundHandler<Object> {
private String wsUri;
private Integer port;
/**
* 描述：读取完连接的消息后，对消息进行处理。
* 这里仅处理HTTP请求，WebSocket请求交给下一个处理器。
*/
@Override
protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
if (msg instanceof FullHttpRequest) {
//判断第一次连接，进行权限校验
authCheck(ctx, (FullHttpRequest) msg);
ctx.fireChannelRead(((FullHttpRequest)msg).retain());
} else if (msg instanceof WebSocketFrame) {
ctx.fireChannelRead(((WebSocketFrame) msg).retain());
}
}
/**
* 第一次HTTP连接的时候，做权限校验，并保存对应的channel,
*/
private void authCheck(ChannelHandlerContext ctx, FullHttpRequest req) {
if (!req.uri().startsWith(wsUri)) {
return;
}
//根据uri中的值来获取值，ws/tocken/MAC
String[] split = req.uri().split("/");
if (split.length < 2) {
return;
}
//通过token做权限校验，并获取用户信息
SysUserInfo userInfo;
try {
userInfo = getUserInfo(split[2]);
} catch (Exception e) {
LoggerUtils.error("连接异常", e);
//返回断开连接，没有权限code信息
HttpResponse response = new DefaultHttpResponse(req.protocolVersion(), HttpResponseStatus.OK);
response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/html; charset=UTF-8");
response.setStatus(HttpResponseStatus.UNAUTHORIZED);
ctx.write(response);
return;
}
//把对应关系放入redis，（channelID , userID）,
//（userID, IP+PORT, Set<channelID>）
//....
//根据缓存中信息，判断每个用户超过最大连接数量 size
//...
LoggerUtils.info(" IP+PORT" + sendServerApi);
//缓存到每次map中 (保存对应channelID和，channel的关系，这样保存也为了删除可以快速删除，不用遍历)
NettyWebSocketHolder.put(ctx.channel().id().asLongText(), ctx.channel());
}
private SysUserInfo getUserInfo(String token) {
//根据自己系统做权限校验
}
/**
* 描述：异常处理，关闭channel
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}
业务处理
1、根据用户的心跳延长对应有redis有效期
2、断开连接清除对应channel信息（1，本地的channel对象；2、redis的channel的ID）
//处理文本协议数据，处理TextWebSocketFrame类型的数据，websocket专门处理文本的frame就是TextWebSocketFrame
public class TextWebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
private RedissonClient redissonClient;
private Integer port;
public TextWebSocketFrameHandler(RedissonClient redissonClient, Integer port) {
this.redissonClient = redissonClient;
this.port = port;
}
//读到客户端的内容并且向客户端去写内容
@Override
protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
LoggerUtils.info("收到消息：" + msg.text());
//接收客户端保持心跳， 延长redis的有效时间
}
//每个channel都有一个唯一的id值
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
//打印出channel唯一值，asLongText方法是channel的id的全名
LoggerUtils.info("handlerAdded：" + ctx.channel().id().asLongText());
}
@Override
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
//断开连接时，删除本地缓存的map中的channel
NettyWebSocketHolder.remove(ctx.channel().id().asLongText());
LoggerUtils.error("handlerRemoved：" + ctx.channel().id().asLongText());
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
//异常断开连接时，删除本地缓存的map中的channel
NettyWebSocketHolder.remove(ctx.channel().id().asLongText());
LoggerUtils.info("异常发生");
ctx.close();
}
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt instanceof IdleStateEvent){
IdleStateEvent idleStateEvent = (IdleStateEvent) evt ;
LoggerUtils.info("userEventTriggered");
if (idleStateEvent.state() == IdleState.READER_IDLE){
LoggerUtils.info("已经很久没有收到信息！");
NettyWebSocketHolder.remove(ctx.channel().id().asLongText());
//清除redis的连接
// ...
LoggerUtils.info("清除后的连接数量："+mapCache.get(sendServerApi));
}
}
super.userEventTriggered(ctx, evt);
}
}
对于nginx的配置
可以利用nginx进行负载均衡，也使用了改域名的ssl加密。在连接的时候需要使用 wss://
# 实际websocket服务器地址
upstream wss_svr {
server 10.186.135.6:22222 weight=1;
server 10.186.135.7:22222 weight=1;
}
# 443 ssl端口配置，实际websockets（wss）地址
server {
listen 443;
server_name *.test.com;
ssl on;
# 自己的证书，放在与nginx.conf同一文件夹下。（若放不同文件夹注意路径问题）
#ssl_protocols TLSv1 TLSv1.1 TLSv1.2;
ssl_certificate /usr/local/zz/nginx/server.crt;
ssl_certificate_key /usr/local/zz/nginx/server.key;
ssl_session_timeout 5m;
ssl_session_cache shared:SSL:10m;
ssl_protocols TLSv1 TLSv1.1 TLSv1.2 SSLv2 SSLv3;
ssl_ciphers ALL:!ADH:!EXPORT56:RC4+RSA:+HIGH:+MEDIUM:+LOW:+SSLv2:+EXP;
ssl_prefer_server_ciphers on;
ssl_verify_client off;
location /ws {
proxy_redirect off;
proxy_pass http://wss_svr; # 转发
proxy_set_header Host $host;
proxy_set_header X-Real_IP $remote_addr;
proxy_set_header X-Forwarded-For $remote_addr:$remote_port;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade; # 升级协议头
proxy_set_header Connection upgrade;
}
}
总结
目前系统服务以及集成设备同步进度的消息推送，与客户端同步稳定，在测试环节有3W台设备的压力测试，保证同步的稳定性。
目前准备接入web页面与pc客户端的二维码登录，通过websocket连接可以保证客户端对消息的实时性变化。
websocket技术可以提升效率，适合对于实时性要求比较高的业务使用。
