基于netty的websocket的生产实践

2019-04-11  本文已影响0人  IT菜鸟学习

问题：由于项目需要与web端进行通讯，web端需要获取同步信息的进度（由于同步过程可能会持续比较长时间，所以采用websocket进行通讯，不在需要web端定时轮询pull拉取进度信息）；

背景：

由于项目需要与web端进行通讯，web端需要获取同步信息的进度（由于同步过程可能会持续比较长时间，所以采用websocket进行通讯，不在需要web端定时轮询pull拉取进度信息）；

解决思路：

  1. 从整体上看是有两个思路：
    前端通过定时器pull去服务端拉取同步信息，通过前端轮询会增加很多无效的请求，即是无效的查询，增加了很多无效的轮询；减少了工作量，加快了开发进度，但是进度同步给用户一个很不实时与直观的界面。
    使用websocket，客户端与服务端增加socket通讯的，所以决定采用push的方式，与web端保持长连接，通过长连接来推送进度消息。
  2. 设备同步流程：
    ~~ 通过前端轮询会增加很多无效的请求，即是无效的查询，增加了很多无效的轮询；所以决定采用push的方式，与web端保持长连接，通过长连接来推送进度消息。

~~ 服务消息是通过web端发起http同步请求，建立socket连接，服务端进行异步处理需要同步的信息，在服务端进行同，通过Kafka发送进度消息，推送到对应消息中转服务，消息中转服务进行 分发到对应的push服务上（netty服务），netty服务上收到对应的消息，通过缓存对应用户的channl进行channel.writeAndFlush()把消息推送的到对应的长连接客户端上，完成整个流程。

image.png

方案介绍

~~对于长连接方案有很对种实现方式，当时也对于选型比较纠结，最终由于与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
image01.png
2. 推送服务会在业务服务端与netty之间建立一个分发机制，分发机制主要作用：

从redis中获取连接通道与netty服务的对应关系，就是对应通道推送到哪一台服务器上。
接收消息与分发消息（通过异步http分发或者kafka分发）。

3. 选型netty的原因，
  1. 由于对应社区的活跃度高；
  2. 服务相对稳定；
  3. 被大多数公司采纳，已经有一套完整的解决方案。
    netty是具有异步非阻塞的特性，传统的IO是面向流的，NIO是面向缓冲区的，是非阻塞的。

netty使用的是reactor模型，有boss线程和worker线程。

服务中的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技术可以提升效率，适合对于实时性要求比较高的业务使用。

上一篇下一篇

猜你喜欢

热点阅读