栏目分类:
子分类:
返回
名师互学网用户登录
快速导航关闭
当前搜索
当前分类
子分类
实用工具
热门搜索
名师互学网 > IT > 软件开发 > 后端开发 > Java

Netty学习四-WebSocket协议开发

Java 更新时间: 发布时间: IT归档 最新发布 模块sitemap 名妆网 法律咨询 聚返吧 英语巴士网 伯小乐 网商动力

Netty学习四-WebSocket协议开发

1 HTTP 协议弊端
  • HTTP 协议为半双工协议,半双工协议是指数据可以在客户端和服务端两个方向传输,但是不能同时传输。意味着同一时刻只有一个方向上的数据传送
  • HTTP 消息冗长且繁琐,HTTP 消息包含消息头、消息体、换行符等,通常情况下采用文本方式传输,相对于其他二进制协议效率低下
  • 实时性不高

没有 WebSocket 之前,很多网站进行消息推送都是客户端轮询,服务端返回最新的数据给客户端。这种方式浏览器需要不停地向服务端发送请求,HTTP 的 header 数据比较冗长,有用数据占用比较少,浪费带宽资源

2 WebSocket 协议特点
  • 单一的 TCP 连接,采用全双工通信模式
  • 对代理、防火墙、路由器透明
  • 无头部信息、Cookie和身份验证
  • 无安全开销
  • 通过 ping/pong 保持链路激活
  • 服务器可以主动给客户端推消息而不用客户端轮询
3 WebSocket 协议开发 3.1 WebSocketHandler 编写
@Slf4j
public class WebSocketHandler extends SimpleChannelInboundHandler {

    private WebSocketServerHandshaker handshaker;

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof IdleStateEvent) {
            log.info("idle connect");
            ctx.writeAndFlush(new TextWebSocketFrame("该连接长时间不活跃,即将关闭"));
            ctx.close();
        } else {
            super.userEventTriggered(ctx, msg);
        }
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof FullHttpRequest) {
            // 传统http接入。第一次请求由 HTTP 承载,
            handleHttpRequest(ctx, (FullHttpRequest) msg);
        } else if (msg instanceof WebSocketFrame) {
            // ws接入
            handleWebSocketFrame(ctx, (WebSocketFrame) msg);
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
        cause.printStackTrace();
    }

    private void handleWebSocketFrame(ChannelHandlerContext context, WebSocketFrame frame) {
        // 判断是否是关闭链路指令
        if (frame instanceof CloseWebSocketFrame) {
            log.info("websocket close");
            handshaker.close(context.channel(), (CloseWebSocketFrame) frame.retain());
            return;
        }
        if (frame instanceof PingWebSocketFrame) {
            log.info("websocket receive msg:{}", "ping");
            context.channel().writeAndFlush(new PongWebSocketFrame(frame.content().retain()));
            return;
        }

        if (!(frame instanceof TextWebSocketFrame)) {
            log.warn("websocket not support binary msg");
            throw new UnsupportedOperationException("不支持二进制");
        }

        String message = ((TextWebSocketFrame) frame).text();
        System.out.println("websocket receive msg:" + message);
        context.channel().writeAndFlush(new TextWebSocketFrame("welcome to netty websocket, currentTime:" + DateUtil.datetimeToString(new Date())));
    }

    private void handleHttpRequest(ChannelHandlerContext context, FullHttpRequest request) {
        if (!request.decoderResult().isSuccess() || !StringUtils.equalsIgnoreCase("websocket", request.headers().get("Upgrade"))) {
            sendHttpResponse(context, request, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
        }

        //可以通过url获取其他参数
        WebSocketServerHandshakerFactory factory = new WebSocketServerHandshakerFactory(
                "ws://" + request.headers().get("Host") + "/websocket", null, false
        );
        // 构建握手处理器
        handshaker = factory.newHandshaker(request);
        if (handshaker == null) {
            WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(context.channel());
        }
        //构建握手响应消息返回客户端,同时将 WebSocket 编解码器添加到 ChannelPipeline 中
        handshaker.handshake(context.channel(), request);
    }

    private void sendHttpResponse(ChannelHandlerContext context, FullHttpRequest request, FullHttpResponse response) {
        if (response.status() != HttpResponseStatus.OK) {
            ByteBuf buf = Unpooled.copiedBuffer(response.status().toString(), CharsetUtil.UTF_8);
            response.content().writeBytes(buf);
            buf.release();
            HttpUtil.setContentLength(response, response.content().readableBytes());
        }

        // 如果非Keep-Alive,关闭连接
        ChannelFuture future = context.channel().writeAndFlush(response);
        if (!HttpUtil.isKeepAlive(request) || response.status() != HttpResponseStatus.OK) {
            future.addListener(ChannelFutureListener.CLOSE);
        }
    }
}
 
3.2 启动类 
        NettyUtil.initServer(8080, () -> {
            List handlers = new ArrayList<>();
            // 将请求和应答消息编码或解码为 HTTP 消息
            handlers.add(new HttpServerCodec());
            // 将 HTTP 消息多个部分组合成一个完整的 HTTP 消息
            handlers.add(new HttpObjectAggregator(65536));
            // 支持浏览器和服务端进行 WebSocket 通信
            handlers.add(new ChunkedWriteHandler());
            // 进行设置心跳检测
            handlers.add(new IdleStateHandler(10, 60, 20, TimeUnit.SECONDS));
            handlers.add(new WebSocketHandler());
            return handlers;
        });
转载请注明:文章转载自 www.mshxw.com
我们一直用心在做
关于我们 文章归档 网站地图 联系我们

版权所有 (c)2021-2022 MSHXW.COM

ICP备案号:晋ICP备2021003244-6号