Netty在Java中的实际应用

Netty#

是什么：一个网络编程框架（基于Java NIO 事件驱动异步）异步：Netty 的所有 I/O 操作均为异步 -> 当调用一个方法时，它立即返回一个 ChannelFuture 对象，而操作本身在后台执行。可以通过这个 Future 对象来获取操作完成的通知或等待其完成。

事件驱动：“当数据来了通知我” Netty基于 事件模型 来处理网络事件 -> 当某个事件发生时（如连接建立、数据到达、异常抛出），Netty自动触发相应的事件，调用预先注册的处理器进行处理。这种模型非常高效，可以处理成千上万的并发连接。NIO：Netty的底层完全依赖于 Java NIO -> Java NIO（New Input/Output）是一个I/O API，它提供了一种基于缓冲区、非阻塞的I/O操作方式，NIO高效地处理大量的并发连接和数据读写操作。

**非阻塞I/O操作：先返回结果，再工作 -> **当任务来了，它立马告诉它知道了，然后开始工作

**阻塞I/O操作：**传统的I/O模式，线程必须等待操作完成

责任链模式：ChannelPipeline 是责任链模式的完美体现。数据进出会像在一条流水线上一样，依次经过多个 ChannelHandler 的处理。每个 Handler 只关心自己负责的逻辑（如解码、日志、业务处理），实现了功能的解耦和高度可定制化。

作用：让程序在高并发下依然又快又稳

怎么用：搭建Websokect连接

关键组件
1. EventLoopGroup和EventLoop：类似线程组和线程，EventLoopGroup分配 EventLoop 来处理新的连接和已有的连接中的事件。一般会创建BossGroup和WorkerGroup两个EventLoopGroup。 **EventLoop：事件循环(I/O事件调度器) -> **一个不断循环的执行体，处理绑定的Channel。
  
  BossGroup＆WorkerGroup：一个负责接收客户端的连接。另一个处理已被接受的连接的 I/O 操作。
2. ServerBootstrap：服务器端的组装和启动类，负责将所有核心组件组合在一起并启动服务器。
3. Channel：**通道，**表示一个网络连接，负责数据的读写和网络操作。所有 I/O 操作都是通过 Channel 进行的。底层是抽象类，只能被子类实现
4. ChannelPipeline：管道。处理数据的责任链，包含一系列处理器 -> 为Channel提供处理器。类似Stream，拦截流经Channel的进出事件
5. ChannelHandler：管道的事件处理器。类似Filter处理 Channel 上各种事件。常见处理器：HttpServerCodec，HttpObjectAggregator
  
  用法：pipeline.addLast(new HttpServerCodec());
6. ChannelFuture：异步操作结果容器，代表一个尚未完成的异步操作。提供查询操作状态的方法
（isDone(), isSuccess()）。I/O 操作都会立即返回一个 ChannelFuture

实际应用

现在要用异步处理WebSoket连接

3.1 添加依赖

1
<dependency>
2
    <groupId>io.netty</groupId>
3
    <artifactId>netty-all</artifactId>
4
    <version>4.1.86.Final</version> <!-- 请使用最新稳定版 -->
5
</dependency>

3.2 创建服务器启动类-通常在websoket-netty目录下

①设置属性

1
implements Runnable：异步，在单独线程中运行
2

3
appConfig：加载配置文件，设置了WS的端口号
4
handlerWebSocket：处理中心
5
handlerTokenValidation：检验TOKEN
6

7
事件调度器组
8
bossGroup：处理连接 -> 老板，接收业务
9
workerGroup：处理消息 -> 员工，处理业务
10

11
@Component
12
@Slf4j
13
public class NettyWebSocketStarter implements Runnable {
14

15
    @Resource
16
    private AppConfig appConfig;
17
    @Resource
18
    private HandlerWebSocket handlerWebSocket;
19
    @Resource
20
    private HandlerTokenValidation handlerTokenValidation;
21

22
    private EventLoopGroup bossGroup = new NioEventLoopGroup();
23
    private EventLoopGroup workerGroup = new NioEventLoopGroup();

1
package com.easymeeting.entity.config;
2

3
@Component("appConfig")
4
public class AppConfig {
5
    private static final Logger logger = LoggerFactory.getLogger(AppConfig.class);
6

7
    /**
8
     * websocket 端口
9
     */
10
    @Value("${ws.port:}")
11
    private Integer wsPort;
12

13
    public Integer getWsPort() {
14
        return wsPort;
15
    }
16
}

③创建启动方法，先开店(ServerBootstrap)，再接客(Channel)。channel创建后才会执行serverBootstrap.channel

1
ServerBootstrap serverBootstrap：创建服务器启动引导类 -> 公司创业指导手册，告诉你怎么开这家快递公司
2
serverBootstrap.group(bossGroup, workerGroup)：设置线程组
3

4
.channel(NioServerSocketChannel.class)：设置通道类型 -> 选择用最高效的NIO运输通道
5
.handler(new LoggingHandler(LogLevel.DEBUG))：处理服务器本身事件
6
.childHandler(new ChannelInitializer<Channel>() {}：处理客户端连接事件
7

8
ChannelInitializer<Channel>()：匿名内部类，需要实现初始化initChannel
9

10
ChannelPipeline：流水线传送带
11

12
@Override
13
public void run() {
14

15
    try {
16
        // 1、
17
        ServerBootstrap serverBootstrap = new ServerBootstrap();
18
        serverBootstrap.group(bossGroup, workerGroup);
19

20
        serverBootstrap
21
                .channel(NioServerSocketChannel.class)
22
                .handler(new LoggingHandler(LogLevel.DEBUG))
23
                .childHandler(new ChannelInitializer<Channel>() {
24

25
                    @Override
26
                    protected void initChannel(Channel channel) {
27

28
                       // 3、
29
                       ChannelPipeline pipeline = channel.pipeline();
30
                       核心，这里写处理逻辑
31

32
                    }
33
                });
34

35
        // 2、
36
        Channel channel=serverBootstrap.bind(appConfig.getWsPort()).sync().channel();
37
        log.info("Netty服务端启动成功,端口:{}", appConfig.getWsPort());
38
        // 阻塞主线程，防止服务器提前退出，等待关闭信号
39
        channel.closeFuture().sync();
40

41
    } catch (Exception e) {
42

43
        log.error("netty启动失败", e);
44

45
    // 优雅关闭，确保所有任务完成后再释放资源
46
    } finally {
47

48
        bossGroup.shutdownGracefully();
49
        workerGroup.shutdownGracefully();
50
    }

④关闭方法

1
@PreDestroy：Spring容器销毁前执行
2
shutdownGracefully()：处理完业务再关闭
3

4
@PreDestroy
5
public void close() {
6
    bossGroup.shutdownGracefully();
7
    workerGroup.shutdownGracefully();
8
}

⑤核心管道配置

1
HttpServerCodec：解码器
2
HttpObjectAggregator：聚合器
3
IdleStateHandler ：心跳监听 -> 看看连接断没断，下面要结合自己的类（HandlerHeartBeat）使用
4
HandlerHeartBeat ：处理心跳结果 -> 要继承“ChannelDuplexHandler”类，这是个Netty的超级Handler类，所有Handler方法都要继承
5

6

7
.childHandler(new ChannelInitializer<Channel>() {
8
    @Override
9
    protected void initChannel(Channel channel) {
10

11
        ChannelPipeline pipeline = channel.pipeline();
12
        /**
13
         * 对http协议的支持，使用http的编码器，解码器
14
         * 通常作为第一个处理器添加
15
         * 必须在 HttpObjectAggregator 之前
16
         */
17
        pipeline.addLast(new HttpServerCodec());
18
        /**
19
         * 这是一个 HTTP 消息聚合器，主要功能是将分片的 HTTP 消息
20
         * （如 chunked 传输编码的消息）聚合成完整的 FullHttpRequest 或 FullHttpResponse。
21
         */
22
        pipeline.addLast(new HttpObjectAggregator(64 * 1024));
23
        /**
24
         * 检测连接空闲状态的处理器 会传递给下一个处理器
25
         * @param readerIdleTime  一段时间内未收到客户端数据。
26
         * @param writerIdleTime  一段时间内未向客户端发送数据。
27
         * @param allIdleTime  -读和写均无活动
28
         */
29
        pipeline.addLast(new IdleStateHandler(10, 0, 0, TimeUnit.SECONDS));
30
        /**
31
         * 处理空闲事件
32
         */
33
        pipeline.addLast(new HandlerHeartBeat());
34
        /**
35
         * 拦截 channelRead 事件
36
         * TOKEN校验
37
         */
38
        pipeline.addLast(handlerTokenValidation);
39
        /**
40
         * WebSocket协议处理器（关键配置）39
41
         * websocketPath 指定 WebSocket 的端点路径
42
         * subprotocols 指定支持的子协议
43
         * allowExtensions  是否允许 WebSocket 扩展
44
         * maxFrameSize  设置最大帧大小 65536
45
         * allowMaskMismatch 是否允许掩码不匹配
46
         * checkStartsWith 是否严格检查路径开头
47
         * handshakeTimeoutMillis  握手超时时间（毫秒）
48
         */
49
        pipeline.addLast(new WebSocketServerProtocolHandler("/ws", null, true, 65536, true, true, 10000L));
50
        pipeline.addLast(handlerWebSocket);
51
    }
52
});

⑥自定的拦截

1
package com.easymeeting.websocket.netty;
2

3
import io.netty.channel.ChannelDuplexHandler;
4
import io.netty.channel.ChannelHandlerContext;
5
import io.netty.handler.timeout.IdleState;
6
import io.netty.handler.timeout.IdleStateEvent;
7
import io.netty.util.Attribute;
8
import io.netty.util.AttributeKey;
9
import org.slf4j.Logger;
10
import org.slf4j.LoggerFactory;
11

12
public class HandlerHeartBeat extends ChannelDuplexHandler {
13
    private static final Logger logger = LoggerFactory.getLogger(HandlerHeartBeat.class);
14

15
    /**
16
     * 处理空闲事件，判断是否是 读取 空闲事件或 写入 空闲事件
17
     * @param ctx 上下文：ChannelHandlerContext
18
     * @param evt 事件：IdleStateEvent
19
     */
20
    // IdleStateEvent 是 Netty 提供的事件类，用于表示通道的空闲状态事件。
21
    // 当通道在指定的时间间隔内没有进行读取或写入操作时，就会触发空闲事件。
22
    // 这里的参数由 Netty 自动传入的
23
    @Override
24
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
25

26
        // 判断是否是空闲事件
27
        if (evt instanceof IdleStateEvent) {
28
            IdleStateEvent e = (IdleStateEvent) evt;
29
            // 判断是否是 读取 空闲事件
30
            if (e.state() == IdleState.READER_IDLE) {
31

32
                // 从channel的属性中获取用户id
33
                Attribute<String> attribute = ctx.channel().attr(AttributeKey.valueOf(ctx.channel().id().toString()));
34
                String userId = attribute.get();
35

36
                // 关闭连接
37
                logger.info("用户{}没有发送心跳断开连接", userId);
38
                ctx.close();
39
            // 判断是否是 写入 空闲事件
40
            } else if (e.state() == IdleState.WRITER_IDLE) {
41
                // 发送心跳包
42
                ctx.writeAndFlush("heart");
43
            }
44
        }
45
    }
46
}

1
package com.easymeeting.websocket.netty;
2

3
@ChannelHandler.Sharable
4
@Component
5
@Slf4j
6
public class HandlerTokenValidation extends SimpleChannelInboundHandler<FullHttpRequest> {
7

8
    @Resource
9
    private ChannelContextUtils channelContextUtils;
10

11
    @Resource
12
    private RedisComponet redisComponet;
13

14
    @Override
15
    protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) {
16
        String uri = request.uri();
17
        QueryStringDecoder queryDecoder = new QueryStringDecoder(uri);
18
        List<String> tokens = queryDecoder.parameters().get("token");
19
        List<String> reconnect = queryDecoder.parameters().get("reconnect");
20
        if (tokens == null) {
21
            sendErrorResponse(ctx);
22
            return;
23
        }
24
        String token = tokens.get(0);
25
        //这里是优化后的重连机制，ws断开后会将用户的会议信息清除，所以重连后从临时信息中获取token信息
26
        TokenUserInfoDto tokenUserInfoDto = null;
27
        if (reconnect != null && Boolean.parseBoolean(reconnect.get(0))) {
28
            log.info("重连中。。。");
29
            //判断临时token信息
30
            tokenUserInfoDto = redisComponet.getTokenUserInfoDtoFromTemp(token);
31
            if (tokenUserInfoDto != null) {
32
                //如果临时缓存有，说明是重连，设置到正式缓存信息中
33
                redisComponet.saveTokenUserInfoDto(tokenUserInfoDto);
34
            }
35
        } else {
36
            tokenUserInfoDto = checkToken(token);
37
        }
38
        if (tokenUserInfoDto == null) {
39
            log.error("校验token失败:{}", token);
40
            sendErrorResponse(ctx);
41
            return;
42
        }
43
        // 如果需要转发消息  增加引用计数
44
        ctx.fireChannelRead(request.retain());
45
        //加入通道
46
        channelContextUtils.addContext(tokenUserInfoDto.getUserId(), ctx.channel());
47
    }
48

49
    private TokenUserInfoDto checkToken(String token) {
50
        if (StringTools.isEmpty(token)) {
51
            return null;
52
        }
53
        TokenUserInfoDto tokenUserInfoDto = redisComponet.getTokenUserInfoDto(token);
54
        return tokenUserInfoDto;
55
    }
56

57
    private void sendErrorResponse(ChannelHandlerContext ctx) {
58
        FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.FORBIDDEN, Unpooled.copiedBuffer("token无效", CharsetUtil.UTF_8));
59
        response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain; charset=UTF-8");
60
        response.headers().set(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());
61
        ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
62
    }
63
}

1
package com.easymeeting.websocket.netty;
2

3

4
/**
5
 * 设置通道共享
6
 */
7
@ChannelHandler.Sharable
8
@Component("handlerWebSocket")
9
public class HandlerWebSocket extends SimpleChannelInboundHandler<TextWebSocketFrame> {
10

11
    private static final Logger logger = LoggerFactory.getLogger(HandlerWebSocket.class);
12

13
    @Resource
14
    private MeetingInfoService meetingInfoService;
15

16
    @Resource
17
    private RedisComponet redisComponet;
18

19
    @Resource
20
    private MessageHandler messageHandler;
21

22
    /**
23
     * 当通道就绪后会调用此方法，通常我们会在这里做一些初始化操作
24
     *
25
     * @param ctx
26
     * @throws Exception
27
     */
28
    @Override
29
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
30
        // Channel channel = ctx.channel();
31
        logger.info("有新的连接加入。。。");
32
    }
33

34
    /**
35
     * 当通道不再活跃时（连接关闭）会调用此方法，我们可以在这里做一些清理工作
36
     *
37
     * @param ctx
38
     * @throws Exception
39
     */
40
    @Override
41
    public void channelInactive(ChannelHandlerContext ctx) {
42
        logger.info("有连接已经断开。。。");
43
        meetingInfoService.removeContext(ctx.channel());
44
    }
45

46
    /**
47
     * 读就绪事件 当有消息可读时会调用此方法，我们可以在这里读取消息并处理。
48
     *
49
     * @param ctx
50
     * @param textWebSocketFrame
51
     * @throws Exception
52
     */
53
    @Override
54
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame textWebSocketFrame) {
55
        //接收心跳
56
        //logger.info("收到消息:{}", textWebSocketFrame.text());
57
        String text = textWebSocketFrame.text();
58
        if (Constants.PING.equals(text)) {
59
            Channel channel = ctx.channel();
60
            Attribute<String> attribute = channel.attr(AttributeKey.valueOf(channel.id().toString()));
61
            String userId = attribute.get();
62
            redisComponet.saveUserHeartBeat(userId);
63
            return;
64
        }
65

66
        PeerConnectionDataDto dataDto = JsonUtils.convertJson2Obj(text, PeerConnectionDataDto.class);
67
        TokenUserInfoDto tokenUserInfoDto = redisComponet.getTokenUserInfoDto(dataDto.getToken());
68
        if (tokenUserInfoDto == null) {
69
            return;
70
        }
71
        MessageSendDto messageSendDto = new MessageSendDto();
72
        messageSendDto.setMessageType(MessageTypeEnum.PEER.getType());
73

74
        PeerMessageDto peerMessageDto = new PeerMessageDto();
75
        peerMessageDto.setSignalType(dataDto.getSignalType());
76
        peerMessageDto.setSignalData(dataDto.getSignalData());
77

78
        messageSendDto.setMessageContent(peerMessageDto);
79
        messageSendDto.setMeetingId(tokenUserInfoDto.getCurrentMeetingId());
80
        messageSendDto.setSendUserId(tokenUserInfoDto.getUserId());
81
        messageSendDto.setReceiveUserId(dataDto.getReceiveUserId());
82
        messageSendDto.setMessageSend2Type(MessageSend2TypeEnum.USER.getType());
83
        messageHandler.sendMessage(messageSendDto);
84
    }
85
}

⑦完整实现

1
package com.easymeeting.websocket.netty;
2

3
/**
4
 * @Description: ws初始化类
5
 */
6
@Component
7
@Slf4j
8
public class NettyWebSocketStarter implements Runnable {
9

10
    @Resource
11
    private AppConfig appConfig;
12

13
    @Resource
14
    private HandlerWebSocket handlerWebSocket;
15

16
    @Resource
17
    private HandlerTokenValidation handlerTokenValidation;
18
    /**
19
     * boss线程组，用于处理连接
20
     */
21
    private EventLoopGroup bossGroup = new NioEventLoopGroup();
22
    /**
23
     * work线程组，用于处理消息
24
     */
25
    private EventLoopGroup workerGroup = new NioEventLoopGroup();
26
    /**
27
     * 资源关闭 -> 在容器销毁时关闭线程组
28
     */
29
    @PreDestroy
30
    public void close() {
31
        bossGroup.shutdownGracefully();
32
        workerGroup.shutdownGracefully();
33
    }
34

35
    @Override
36
    public void run() {
37
        try {
38
            //创建服务端启动助手
39
            ServerBootstrap serverBootstrap = new ServerBootstrap();
40
            // 添加线程组
41
            serverBootstrap.group(bossGroup, workerGroup);
42
            // 添加通道工厂
43
            serverBootstrap.channel(NioServerSocketChannel.class)
44
                    // 添加通道初 始化器
45
                    .handler(new LoggingHandler(LogLevel.DEBUG))
46
                    // 添加子通道初 始化器
47
                    .childHandler(new ChannelInitializer<Channel>() {
48
                        // 初始化子通道（处理器）
49
                        @Override
50
                        protected void initChannel(Channel channel) {
51

52
                            ChannelPipeline pipeline = channel.pipeline();
53
                            /**
54
                             * 对http协议的支持，使用http的编码器，解码器
55
                             * 通常作为第一个处理器添加
56
                             * 必须在 HttpObjectAggregator 之前
57
                             */
58
                            pipeline.addLast(new HttpServerCodec());
59
                            /**
60
                             * 这是一个 HTTP 消息聚合器，主要功能是将分片的 HTTP 消息
61
                             * （如 chunked 传输编码的消息）聚合成完整的 FullHttpRequest 或 FullHttpResponse。
62
                             */
63
                            pipeline.addLast(new HttpObjectAggregator(64 * 1024));
64
                            /**
65
                             * 检测连接空闲状态的处理器 会传递给下一个处理器
66
                             * @param readerIdleTime  一段时间内未收到客户端数据。
67
                             * @param writerIdleTime  一段时间内未向客户端发送数据。
68
                             * @param allIdleTime  -读和写均无活动
69
                             */
70
                            pipeline.addLast(new IdleStateHandler(10, 0, 0, TimeUnit.SECONDS));
71
                            /**                                                                                                                               vcx
72
                             * 处理空闲事件
73
                             */
74
                            pipeline.addLast(new HandlerHeartBeat());
75
                            /**
76
                             * 拦截 channelRead 事件
77
                             * TOKEN校验
78
                             */
79
                            pipeline.addLast(handlerTokenValidation);
80
                            /**
81
                             * WebSocket协议处理器（关键配置）39
82
                             * websocketPath 指定 WebSocket 的端点路径
83
                             * subprotocols 指定支持的子协议
84
                             * allowExtensions  是否允许 WebSocket 扩展
85
                             * maxFrameSize  设置最大帧大小 65536
86
                             * allowMaskMismatch 是否允许掩码不匹配
87
                             * checkStartsWith 是否严格检查路径开头
88
                             * handshakeTimeoutMillis  握手超时时间（毫秒）
89
                             */
90
                            pipeline.addLast(new WebSocketServerProtocolHandler("/ws", null, true, 65536, true, true, 10000L));
91
                            pipeline.addLast(handlerWebSocket);
92
                        }
93
                    });
94
            //启动
95
            Channel channel = serverBootstrap.bind(appConfig.getWsPort()).sync().channel();
96
            log.info("Netty服务端启动成功,端口:{}", appConfig.getWsPort());
97
            channel.closeFuture().sync();
98
        } catch (Exception e) {
99
            log.error("netty启动失败", e);
100
        } finally {
101
            // 关闭线程组
102
            bossGroup.shutdownGracefully();
103
            workerGroup.shutdownGracefully();
104
        }
105
    }
106
}

3.3 创建另一个启动类，用于异步启动其它服务 -> 与启动类在同一目录，异步启动创建netty

1
package com.easymeeting;
2
/**
3
 * 异步初始化服务
4
 */
5
@Component("initRun")
6
public class InitRun implements ApplicationRunner {
7

8
    @Resource
9
    private NettyWebSocketStarter nettyWebSocketStarter;
10

11
    @Override
12
    public void run(ApplicationArguments args) {
13

14
        //启动Neety
15
        new Thread(nettyWebSocketStarter).start();
16

17
    }
18
}

Love•Life

Netty#