使用netty开启websocket通信收取消息

前言

这是我在这个网站整理的笔记,有错误的地方请指出，关注我，接下来还会持续更新。

作者：神的孩子都在歌唱

Netty 是一个基于 Java 的网络编程框架，专注于提供高性能和可扩展性的网络通信解决方案。其设计目标是简化网络应用程序的开发，并提供高度的定制性和灵活性。

一. Netty主要特性

异步事件驱动：Netty 使用 NIO（非阻塞 I/O）技术，采用了异步事件驱动的编程模型，允许同时处理多个连接和请求。这种机制使得Netty能够高效地处理大量的并发连接。

简单易用的API： Netty提供了简洁、直观的API，降低了网络应用程序的开发难度。它的组件化架构使得开发人员可以轻松地组合和重用各种组件来构建自己的网络应用。

高性能和可扩展性： 通过优化的IO处理和并发模型，Netty能够提供高性能的网络通信能力。其可扩展的架构使得可以根据应用需求进行灵活扩展，应对不同规模的网络需求。

多协议支持： Netty支持多种网络协议，包括但不限于HTTP、WebSocket、TCP和UDP。这使得它适用于构建各种类型的网络应用程序，从简单的HTTP服务器到实时的多用户聊天应用。

二. 代码示例

2.1 WebSocket服务器启动类

/**
 * @author chenyunzhi
 * @date 2023/12/26 16:36
 * @Description  AIO模型 netty启动类
 */
public class WebSocketServer {
​
    public void run() {
        // 主线程池
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        // 从线程池
        EventLoopGroup subGroup = new NioEventLoopGroup();
        try {
            // 启动类
            ServerBootstrap server = new ServerBootstrap()
                    // 指定bossGroup和subGroup作为EventLoopGroup，用于处理客户端连接和客户端I/O操作
                    .group(bossGroup, subGroup)
                    // 指定NioServerSocketChannel作为服务器的Channel类型 网络传输
                    .channel(NioServerSocketChannel.class)
                    // 指定ChannelHandlerInitializer作为处理新连接的ChannelHandler
                    .childHandler(new ChannelHandlerInitializer())
                    // 设置服务器套接字的连接队列大小为128
                    .option(ChannelOption.SO_BACKLOG, 128)
                    // 设置子通道的SO_KEEPALIVE选项为true，以保持长连接
                    .childOption(ChannelOption.SO_KEEPALIVE, true);;
            // 启动连接
            server.bind(8088).sync().channel().closeFuture().sync();;
            System.out.println("netty websocket server 启动完毕...");
        } catch (Exception e) {
            System.out.println(e.getMessage());
        } finally {
            bossGroup.shutdownGracefully();
            subGroup.shutdownGracefully();
        }
    }
    public static void main(String[] args) {
        new WebSocketServer().run();
    }
}

在这个类中：

• EventLoopGroup是Netty中用于处理事件循环的线程池，bossGroup用于接受连接请求，subGroup用于处理连接的I/O操作。

• ServerBootstrap是Netty提供的用于设置服务器的启动辅助类，配置服务器参数、通道类型和处理器。

• channel()指定了使用NIO进行网络传输，childHandler()设置了用于处理新连接的处理器。

• option()和childOption()分别用于设置服务器套接字和子通道的选项。

• bind()方法绑定了服务器端口并开始接收连接请求，sync()方法会阻塞等待服务器socket关闭。

我们自定义了一个通道处理类ChannelHandlerInitializer

2.2 自定义通道处理类

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
​
/**
 * @author chenyunzhi
 * @date 2023/12/26 16:46
 * @Description
 */
public class ChannelHandlerInitializer extends ChannelInitializer<SocketChannel> {
​
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        // 获取管道（pipline）
        ChannelPipeline pipeline = ch.pipeline();
​
        // websocket基于http协议,所以要有http编码器
        pipeline.addLast(new HttpServerCodec());
​
        // 添加支持异步大文件传输的处理器
        pipeline.addLast(new ChunkedWriteHandler());
​
        // 添加对HTTP消息的聚合处理，最大消息大小为64KB
        pipeline.addLast(new HttpObjectAggregator(1024 * 64));
​
        // 添加WebSocket协议处理器，指定WebSocket的访问路径
        pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
​
        // 自定义的handler，处理实时消息
        pipeline.addLast(new ChatHandler());
    }
}

这里我们：

• 继承了ChannelInitializer<SocketChannel>，这个类用于初始化SocketChannel的管道。

• initChannel()方法会在Channel被注册时调用，我们在这里配置了处理器。

• ChannelPipeline是一个处理器链，通过addLast()方法将编解码器和处理器按顺序添加到管道中。

最后，我们来看自定义的消息处理管道类：

2.3 自定义消息处理管道类

​
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
​
import java.time.LocalDateTime;
​
/**
 * @author chenyunzhi
 * @date 2023/12/26 16:48
 * @Description 消息处理器 负责处理接收的消息
 */
public class ChatHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
​
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        String string = msg.text();
        System.out.println(string);
        // 将数据刷新到客户端上
        ctx.writeAndFlush(
                new TextWebSocketFrame(
                        "[服务器在]" + LocalDateTime.now()
                                + "接受到消息, 消息为：" + string));
​
    }
}

这个类是用来处理接收到的WebSocket消息的，其中：

• SimpleChannelInboundHandler<TextWebSocketFrame>是一个处理器，用于处理接收到的TextWebSocketFrame类型的消息。

• channelRead0()方法会在接收到消息时被调用，这里处理了接收到的文本消息。

• ChannelHandlerContext表示处理器上下文，可以用来与ChannelPipeline和其他处理器进行交互。

三. 测试

将postman当作客户端发送请求，测试如下

服务端能够正常的接收和返回消息

一些客户端测试工具：

• http://www.jsons.cn/websocket/

作者：神的孩子都在歌唱

本人博客：https://blog.csdn.net/weixin_46654114

转载说明：务必注明来源，附带本人博客连接。