使用 Netty 实现 WebSocket 通信

🕗 发布于 2024-12-11 01:09 websocket 网络协议 netty

WebSocket 是一种全双工通信协议，可以在客户端和服务器之间建立持久连接，广泛应用于即时聊天、实时数据推送等场景。Netty 提供了对 WebSocket 协议的全面支持，可以快速实现一个高效的 WebSocket 服务。

本文将从基础原理出发，逐步讲解如何使用 Netty 实现 WebSocket 通信。

1. WebSocket 基本原理

1.1 什么是 WebSocket？

WebSocket 是一个基于 TCP 的全双工协议，通过一次 HTTP 请求升级为 WebSocket 协议后，客户端和服务器可以在单个连接上自由发送消息。
WebSocket 的默认端口：
- 80（非加密通信）
- 443（加密通信）

1.2 WebSocket 工作流程

连接建立：通过 HTTP 请求，使用 Upgrade: websocket 升级协议。
数据传输：升级后使用帧（Frame）进行数据传输，帧可以是文本、二进制或控制信息。
连接关闭：客户端或服务器可随时关闭连接。

2. Netty 对 WebSocket 的支持

Netty 提供了以下 WebSocket 处理器：

WebSocketServerProtocolHandler：处理 WebSocket 协议升级和帧的管理。
TextWebSocketFrame：表示 WebSocket 文本帧。
BinaryWebSocketFrame：表示 WebSocket 二进制帧。
PingWebSocketFrame 和 PongWebSocketFrame：用于心跳检测。

3. 使用 Netty 实现 WebSocket 服务

以下是完整的实现过程，包括服务端和客户端代码。

3.1 服务端代码

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;

public class WebSocketServer {
    public static void main(String[] args) throws InterruptedException {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                     .channel(NioServerSocketChannel.class)
                     .childHandler(new ChannelInitializer<Channel>() {
                         @Override
                         protected void initChannel(Channel ch) throws Exception {
                             ChannelPipeline pipeline = ch.pipeline();

                             // HTTP 协议支持
                             pipeline.addLast(new HttpServerCodec());
                             pipeline.addLast(new HttpObjectAggregator(65536));
                             pipeline.addLast(new ChunkedWriteHandler());

                             // WebSocket 协议支持
                             pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));

                             // 自定义处理器
                             pipeline.addLast(new WebSocketFrameHandler());
                         }
                     });

            System.out.println("WebSocket server started on port 8080");
            ChannelFuture future = bootstrap.bind(8080).sync();
            future.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

class WebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame frame) throws Exception {
        String receivedText = frame.text();
        System.out.println("Received: " + receivedText);

        // 返回响应
        ctx.channel().writeAndFlush(new TextWebSocketFrame("Server received: " + receivedText));
    }

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        System.out.println("Client connected: " + ctx.channel().id().asLongText());
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        System.out.println("Client disconnected: " + ctx.channel().id().asLongText());
    }

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

服务端代码说明

HTTP 支持：
- HttpServerCodec：HTTP 请求和响应的编解码器。
- HttpObjectAggregator：将多个 HTTP 消息聚合成一个完整的 FullHttpRequest 或 FullHttpResponse。
- ChunkedWriteHandler：支持大数据流的写操作。
WebSocket 支持：
- WebSocketServerProtocolHandler：
  - 处理 WebSocket 协议的升级（HTTP -> WebSocket）。
  - /ws：指定 WebSocket 的路径。
自定义处理器：
- WebSocketFrameHandler：处理 WebSocket 帧，接收文本消息并发送响应。

3.2 客户端代码

客户端可以使用浏览器或自定义 WebSocket 客户端库（如 JavaScript 的 WebSocket 对象）。

以下是一个 JavaScript 客户端示例：

const ws = new WebSocket('ws://localhost:8080/ws');

// 连接建立
ws.onopen = () => {
    console.log('Connected to server');
    ws.send('Hello Server!');
};

// 接收消息
ws.onmessage = (event) => {
    console.log('Received from server:', event.data);
};

// 连接关闭
ws.onclose = () => {
    console.log('Connection closed');
};

运行代码时，可以在浏览器开发者工具的控制台中看到通信结果。

4. 功能扩展

4.1 广播消息

在服务端实现广播功能，可以向所有连接的客户端发送消息。

修改 WebSocketFrameHandler：

import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;

public class WebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    private static final ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame frame) throws Exception {
        String receivedText = frame.text();
        System.out.println("Received: " + receivedText);

        // 广播消息到所有客户端
        for (Channel channel : channels) {
            channel.writeAndFlush(new TextWebSocketFrame("[Broadcast] " + receivedText));
        }
    }

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        channels.add(ctx.channel());
        System.out.println("Client connected: " + ctx.channel().id().asLongText());
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        channels.remove(ctx.channel());
        System.out.println("Client disconnected: " + ctx.channel().id().asLongText());
    }
}

4.2 心跳检测

通过 IdleStateHandler 实现心跳检测，检测客户端是否存活。

import io.netty.handler.timeout.IdleStateHandler;

@Override
protected void initChannel(Channel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();

    // 添加心跳检测
    pipeline.addLast(new IdleStateHandler(60, 30, 0)); // 读超时 60 秒，写超时 30 秒

    // 其他处理器
    pipeline.addLast(new WebSocketFrameHandler());
}

5. 测试流程

启动服务端代码，监听 8080 端口。
使用浏览器或 WebSocket 客户端连接 ws://localhost:8080/ws。
测试发送和接收消息。
多个客户端连接时，验证广播功能。

6. 总结

通过本文实现了一个简单的 WebSocket 服务，展示了 Netty 对 WebSocket 的强大支持。Netty 提供了对 WebSocket 协议的底层支持，以及灵活的扩展功能，适合构建高性能的实时通信服务。

关键点回顾：

WebSocket 协议升级：通过 WebSocketServerProtocolHandler 实现。
帧的处理：通过 TextWebSocketFrame 和 BinaryWebSocketFrame 等对象实现数据收发。
功能扩展：实现广播消息和心跳检测。

通过这些功能，你可以进一步开发复杂的实时通信系统，如在线聊天室、股票实时推送等应用场景。

原文地址：https://blog.csdn.net/u012561308/article/details/144339717

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：Whisper：Robust Speech Recognition via Large-Scale Weak Supervision
下一篇：python爬虫--小白篇【反反爬】

.NET(C#) 如何配置用户首选项及保存用户设置
.NET(C#) 如何配置用户首选项及保存用户设置
阅读更多2024-12-14
【最新】北大数字普惠金融指数数据集-省市县（2011-2023年）
郭峰,王靖一,王芳,孔涛,张勋,程志云.测度中国数字普惠金融发展:指数编制与空间特征[J].经济学(季刊),2020,19(04):1401-1418.时间跨度：省级和城市级指数时间跨度为2011-2
阅读更多2024-12-14
GESP202412 四级【Recamán】题解（AC）
a11ak−1−kkakak−1−kak−1k小杨想知道 Recamán 数列的前n项从小到大排序后的结果。手动计算非常困难，小杨希望你能帮他解决这个问题。
阅读更多2024-12-14
IDEA遇到EasyConnect中的网络资源无法访问的问题
版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。原文链接：https://blog.csdn.net/wanshanyu_/article/de
阅读更多2024-12-14
双目摄像头标定方法
此时已经完成标定，左下角为反投影误差，右边为外参可视化。将双目左右目拍的图像上传（左右目最好不少于20张）此时回到主页面，即可看到成功导出。把这些误差大的删除即可。
阅读更多2024-12-14
Servlet、omcat服务器架构与工作原理
Servlet是运行在服务器端的Java程序，它的主要职责之一是接收并处理来自客户端（如浏览器）的HTTP请求。当客户端发送一个请求到服务器时，Servlet可以解析请求中的信息，例如请求的URL路径
阅读更多2024-12-14
Vue生命周期钩子函数：深入解析与实践
作为高级Vue前端开发人员，对Vue组件的生命周期钩子函数有着深刻的理解是至关重要的。生命周期钩子函数是指在Vue组件的创建、更新、销毁等过程中，Vue自动调用的一系列方法。通过这些钩子函数，我们可以
阅读更多2024-12-14
安卓开发--使用android studio发布APP
app发布
阅读更多2024-12-14
数据结构与算法学习笔记----拓扑排序
@ author: 明月清了个风。
阅读更多2024-12-14
python 将数据保存到现有的Excel文件的新工作表
out_file = ‘query.xlsx’df1 = pd.DataFrame(out_data)若直接写入：df1.to_excel(out_file, index=False, sheet_n
阅读更多2024-12-14