做暖dnf动态ufo网站,如何用kali做网站渗透,东莞凤岗网站制作,郑州好的建网站公司Netty Netty 的核心概念Netty 的主要特性Netty 的应用场景Netty 的基本使用服务器端处理器 总结 代码分析1.心跳检测代码解析类和成员变量userEventTriggered方法总结 4.参数详解ChannelHandlerContext ctxObject evt 事件来源示例#xff1a;配置 IdleStateHandler事件处理示… Netty Netty 的核心概念Netty 的主要特性Netty 的应用场景Netty 的基本使用服务器端处理器 总结 代码分析1.心跳检测代码解析类和成员变量userEventTriggered方法总结 4.参数详解ChannelHandlerContext ctxObject evt 事件来源示例配置 IdleStateHandler事件处理示例回调 2.Netty WebSocket 服务器启动器类1.代码解析类和成员变量资源关闭方法run 方法 2.总结3.Netty WebSocket 处理器 HandlerWebSocket代码解析类和成员变量channelActive 方法channelInactive 方法channelRead0 方法userEventTriggered 方法getToken 方法 总结 完整过程1. 客户端连接到 WebSocket 服务器2. 服务器端初始化和配置Netty 服务器启动 3. 通道初始化4. 握手和连接事件WebSocket 协议处理和心跳检测心跳检测 5. 消息处理6. 连接关闭总结 Netty 是一个基于 Java 的异步事件驱动的网络应用框架用于快速开发高性能、高可靠性的网络应用。Netty 提供了丰富的 API支持多种传输协议和多种编解码方式广泛应用于高性能的网络服务器和客户端的开发。
Netty 的核心概念 ChannelNetty 中的基本网络操作抽象。它代表一个打开的连接比如一个到远程服务器的 TCP 连接。Channel 提供了异步的网络 I/O 操作如读、写、连接和绑定。 EventLoopNetty 中处理 I/O 操作的核心。EventLoop 是一个处理 I/O 事件、运行任务和处理定时任务的循环。每个 Channel 都会分配给一个 EventLoop。 ChannelFutureNetty 中的异步操作结果。ChannelFuture 提供了在操作完成时通知的机制允许你在操作完成后执行一些特定的操作如写操作完成后的回调处理。 ChannelHandler用于处理 Channel 的 I/O 事件和数据。ChannelHandler 是处理网络事件和数据的核心接口。你可以实现 ChannelHandler 来处理入站和出站数据。 ChannelPipelineNetty 中的责任链模式实现。每个 Channel 都有一个 ChannelPipeline它持有一个 ChannelHandler 的链。当一个 I/O 事件发生时事件在 ChannelPipeline 中传播通过其中的 ChannelHandler 进行处理。 ByteBufNetty 中的数据容器比 Java 的 NIO ByteBuffer 更高效和灵活。ByteBuf 提供了丰富的 API支持动态扩展、零拷贝等特性。
Netty 的主要特性 异步和事件驱动Netty 基于异步和事件驱动的编程模型能够处理大量并发连接并提供高吞吐量和低延迟。 多协议支持Netty 支持多种协议如 HTTP、WebSocket、TCP、UDP 等。你可以轻松实现自定义协议。 高效的内存管理Netty 提供了高效的内存管理机制避免了频繁的垃圾回收提升了应用的性能。 丰富的工具类和扩展点Netty 提供了大量的工具类和扩展点方便开发者定制和扩展。
Netty 的应用场景
高性能的网络服务器如 HTTP 服务器、游戏服务器、聊天服务器等。分布式系统中的网络通信如 RPC 框架、消息中间件等。代理服务器如反向代理、网关等。任何需要高并发和低延迟的网络应用。
Netty 的基本使用
以下是一个简单的 Netty 服务器示例
服务器端
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;public class NettyServer {private final int port;public NettyServer(int port) {this.port port;}public void start() throws Exception {EventLoopGroup bossGroup new NioEventLoopGroup(1); // 处理连接事件EventLoopGroup workerGroup new NioEventLoopGroup(); // 处理I/O事件try {ServerBootstrap b new ServerBootstrap();b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializerSocketChannel() {Overridepublic void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new EchoServerHandler());}}).option(ChannelOption.SO_BACKLOG, 128).childOption(ChannelOption.SO_KEEPALIVE, true);ChannelFuture f b.bind(port).sync(); // 绑定端口f.channel().closeFuture().sync(); // 等待服务器关闭} finally {workerGroup.shutdownGracefully();bossGroup.shutdownGracefully();}}public static void main(String[] args) throws Exception {int port 8080;new NettyServer(port).start();}
}处理器
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;public class EchoServerHandler extends ChannelInboundHandlerAdapter {Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) {ctx.write(msg); // 将接收到的消息写回去}Overridepublic void channelReadComplete(ChannelHandlerContext ctx) {ctx.flush(); // 刷新所有写入到客户端的消息}Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {cause.printStackTrace();ctx.close(); // 发生异常时关闭连接}
}总结
Netty 是一个强大的网络编程框架适用于构建高性能、高可用性的网络应用。它提供了丰富的 API 和灵活的扩展机制使得开发高性能的网络应用变得更加简单和高效。通过异步事件驱动模型和高效的内存管理Netty 可以处理大量并发连接并提供低延迟的服务。 代码分析
1.心跳检测
public class HandlerHeartBeat extends ChannelDuplexHandler {private static final Logger logger LoggerFactory.getLogger(HandlerHeartBeat.class);Overridepublic void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {if (evt instanceof IdleStateEvent) {IdleStateEvent e (IdleStateEvent) evt;if (e.state() IdleState.READER_IDLE) {AttributeString attribute ctx.channel().attr(AttributeKey.valueOf(ctx.channel().id().toString()));String userId attribute.get();logger.info(用户{}没有发送心跳断开连接, userId);ctx.close();} else if (e.state() IdleState.WRITER_IDLE) {ctx.writeAndFlush(heart);}}}
}代码解析
这是一个Netty的处理器类HandlerHeartBeat继承了ChannelDuplexHandler。这个类主要用于处理心跳检测逻辑以确保连接的存活性。以下是对代码的详细解析
类和成员变量
public class HandlerHeartBeat extends ChannelDuplexHandler {private static final Logger logger LoggerFactory.getLogger(HandlerHeartBeat.class);
}HandlerHeartBeat继承自ChannelDuplexHandler它是Netty提供的一个用于处理双向事件的处理器类。
userEventTriggered方法
Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {if (evt instanceof IdleStateEvent) {IdleStateEvent e (IdleStateEvent) evt;if (e.state() IdleState.READER_IDLE) {AttributeString attribute ctx.channel().attr(AttributeKey.valueOf(ctx.channel().id().toString()));String userId attribute.get();logger.info(用户{}没有发送心跳断开连接, userId);ctx.close();} else if (e.state() IdleState.WRITER_IDLE) {ctx.writeAndFlush(heart);}}
}userEventTriggered这是Netty中的一个回调方法当用户事件触发时会调用这个方法。在心跳检测中当连接变为空闲时Netty会触发一个IdleStateEvent事件。 if (evt instanceof IdleStateEvent)检查事件是否为IdleStateEvent。IdleStateEvent是Netty提供的一个特殊事件用于表示连接的空闲状态。 IdleStateEvent e (IdleStateEvent) evt将事件强制转换为IdleStateEvent。 处理READER_IDLE状态 if (e.state() IdleState.READER_IDLE) {AttributeString attribute ctx.channel().attr(AttributeKey.valueOf(ctx.channel().id().toString()));String userId attribute.get();logger.info(用户{}没有发送心跳断开连接, userId);ctx.close();
}IdleState.READER_IDLE表示读取通道空闲即长时间没有从客户端读取到数据。ctx.channel().attr(...)获取与通道相关的属性。在这里通过通道的ID作为键来获取用户ID。logger.info(...)记录日志说明哪个用户没有发送心跳包导致连接断开。ctx.close()关闭连接。 处理WRITER_IDLE状态 else if (e.state() IdleState.WRITER_IDLE) {ctx.writeAndFlush(heart);
}IdleState.WRITER_IDLE表示写入通道空闲即长时间没有向客户端发送数据。ctx.writeAndFlush(heart)向客户端发送一个心跳消息 heart保持连接的活跃状态。
总结
这个处理器类HandlerHeartBeat主要用于处理心跳检测逻辑以确保客户端和服务器之间的连接在长时间没有数据交互时保持活跃或及时关闭
如果长时间没有读取到数据READER_IDLE则关闭连接并记录日志。如果长时间没有向客户端发送数据WRITER_IDLE则发送一个心跳消息 heart 以保持连接活跃。 在Netty中userEventTriggered 方法的参数是 ChannelHandlerContext 和 Object 类型的事件。以下是详细解释
4.参数详解
ChannelHandlerContext ctx
类型ChannelHandlerContext作用ChannelHandlerContext 提供了各种操作以触发 IO 操作和事件处理如读取、写入、连接、断开等。它关联了一个 Channel并且允许访问 ChannelPipeline 中的其他 ChannelHandler。
Object evt
类型Object作用这是传递给该方法的事件对象。在心跳检测的场景下这个事件对象通常是 IdleStateEvent它表示连接的空闲状态读空闲、写空闲、读写空闲。
事件来源
在Netty中空闲状态检测通常是通过 IdleStateHandler 来实现的。IdleStateHandler 会监测通道的读写操作如果通道在指定的时间内没有读或写操作就会触发 IdleStateEvent 事件。
示例配置 IdleStateHandler
以下是一个示例展示如何将 IdleStateHandler 添加到 ChannelPipeline 中以便在通道空闲时触发 IdleStateEvent
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.timeout.IdleStateHandler;public class NettyServer {public static void main(String[] args) throws Exception {NioEventLoopGroup bossGroup new NioEventLoopGroup(1);NioEventLoopGroup workerGroup new NioEventLoopGroup();try {ServerBootstrap b new ServerBootstrap();b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializerSocketChannel() {Overridepublic void initChannel(SocketChannel ch) throws Exception {ChannelPipeline p ch.pipeline();// 添加 IdleStateHandler配置读、写空闲时间p.addLast(new IdleStateHandler(60, 30, 0)); // 读空闲60秒写空闲30秒// 添加自定义的心跳检测处理器p.addLast(new HandlerHeartBeat());}});b.bind(8080).sync().channel().closeFuture().sync();} finally {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}
}事件处理
当 IdleStateHandler 发现通道空闲时会触发 IdleStateEvent并调用 HandlerHeartBeat 中的 userEventTriggered 方法。此时ctx 和 evt 参数的具体信息如下
ctx当前通道的上下文提供了通道的各种操作方法。evt具体的事件对象这里是 IdleStateEvent表示通道的空闲状态。
示例回调
假设配置的读空闲时间是60秒写空闲时间是30秒
如果60秒内没有读取到任何数据IdleStateHandler 会触发 IdleStateEvent其中状态为 IdleState.READER_IDLE。如果30秒内没有向通道写入数据IdleStateHandler 会触发 IdleStateEvent其中状态为 IdleState.WRITER_IDLE。
在 HandlerHeartBeat 中
Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {if (evt instanceof IdleStateEvent) {IdleStateEvent e (IdleStateEvent) evt;if (e.state() IdleState.READER_IDLE) {AttributeString attribute ctx.channel().attr(AttributeKey.valueOf(ctx.channel().id().toString()));String userId attribute.get();logger.info(用户{}没有发送心跳断开连接, userId);ctx.close();} else if (e.state() IdleState.WRITER_IDLE) {ctx.writeAndFlush(heart);}}
}IdleStateEvent包含了通道的空闲状态读空闲、写空闲、读写空闲。IdleState.READER_IDLE表示读空闲事件。IdleState.WRITER_IDLE表示写空闲事件。
2.Netty WebSocket 服务器启动器类
Component
public class NettyWebSocketStarter implements Runnable {private static final Logger logger LoggerFactory.getLogger(NettyWebSocketStarter.class);Resourceprivate AppConfig appConfig;Resourceprivate HandlerWebSocket handlerWebSocket;/*** boss线程组用于处理连接*/private EventLoopGroup bossGroup new NioEventLoopGroup(1);/*** work线程组用于处理消息*/private EventLoopGroup workerGroup new NioEventLoopGroup();/*** 资源关闭——在容器销毁时关闭*/PreDestroypublic void close() {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}Overridepublic void run() {try {//创建服务端启动助手ServerBootstrap serverBootstrap new ServerBootstrap();serverBootstrap.group(bossGroup, workerGroup);serverBootstrap.channel(NioServerSocketChannel.class).handler(new LoggingHandler(LogLevel.DEBUG)).childHandler(new ChannelInitializerChannel() {Overrideprotected void initChannel(Channel channel) {ChannelPipeline pipeline channel.pipeline();//设置几个重要的处理器// 对http协议的支持使用http的编码器解码器pipeline.addLast(new HttpServerCodec());//聚合解码 httpRequest/htppContent/lastHttpContent到fullHttpRequest//保证接收的http请求的完整性pipeline.addLast(new HttpObjectAggregator(64 * 1024));//心跳 long readerIdleTime, long writerIdleTime, long allIdleTime, TimeUnit unit// readerIdleTime 读超时事件 即测试段一定事件内未接收到被测试段消息// writerIdleTime 为写超时时间 即测试端一定时间内想被测试端发送消息//allIdleTime 所有类型的超时时间pipeline.addLast(new IdleStateHandler(60, 0, 0, TimeUnit.SECONDS));pipeline.addLast(new HandlerHeartBeat());//将http协议升级为ws协议对websocket支持pipeline.addLast(new WebSocketServerProtocolHandler(/ws, null, true, 64 * 1024, true, true, 10000L));pipeline.addLast(handlerWebSocket);}});//启动ChannelFuture channelFuture serverBootstrap.bind(appConfig.getWsPort()).sync();logger.info(Netty服务端启动成功,端口:{}, appConfig.getWsPort());channelFuture.channel().closeFuture().sync();} catch (Exception e) {e.printStackTrace();bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();} finally {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}
}
1.代码解析
这段代码定义了一个 Netty WebSocket 服务器启动器类 NettyWebSocketStarter该类实现了 Runnable 接口可以在独立的线程中运行。这个类负责配置并启动 Netty WebSocket 服务器。
类和成员变量
Component
public class NettyWebSocketStarter implements Runnable {private static final Logger logger LoggerFactory.getLogger(NettyWebSocketStarter.class);Resourceprivate AppConfig appConfig;Resourceprivate HandlerWebSocket handlerWebSocket;/*** boss线程组用于处理连接*/private EventLoopGroup bossGroup new NioEventLoopGroup(1);/*** work线程组用于处理消息*/private EventLoopGroup workerGroup new NioEventLoopGroup();
}NettyWebSocketStarter实现了 Runnable 接口使得该类可以在独立的线程中运行。logger用于记录日志的 Logger 对象。appConfig注入的应用配置类用于获取配置项如 WebSocket 端口。handlerWebSocket注入的 WebSocket 处理器用于处理 WebSocket 消息。bossGroup 和 workerGroup分别用于处理连接和处理消息的线程组。
资源关闭方法
/*** 资源关闭——在容器销毁时关闭*/
PreDestroy
public void close() {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();
}PreDestroy在 Spring 容器销毁之前调用 close 方法优雅地关闭 bossGroup 和 workerGroup释放资源。
run 方法
Override
public void run() {try {//创建服务端启动助手ServerBootstrap serverBootstrap new ServerBootstrap();serverBootstrap.group(bossGroup, workerGroup);serverBootstrap.channel(NioServerSocketChannel.class).handler(new LoggingHandler(LogLevel.DEBUG)).childHandler(new ChannelInitializerChannel() {Overrideprotected void initChannel(Channel channel) {ChannelPipeline pipeline channel.pipeline();//设置几个重要的处理器// 对http协议的支持使用http的编码器解码器pipeline.addLast(new HttpServerCodec());//聚合解码 httpRequest/htppContent/lastHttpContent到fullHttpRequest//保证接收的http请求的完整性pipeline.addLast(new HttpObjectAggregator(64 * 1024));//心跳 long readerIdleTime, long writerIdleTime, long allIdleTime, TimeUnit unit// readerIdleTime 读超时事件 即测试段一定事件内未接收到被测试段消息// writerIdleTime 为写超时时间 即测试端一定时间内想被测试端发送消息//allIdleTime 所有类型的超时时间pipeline.addLast(new IdleStateHandler(60, 0, 0, TimeUnit.SECONDS));pipeline.addLast(new HandlerHeartBeat());//将http协议升级为ws协议对websocket支持pipeline.addLast(new WebSocketServerProtocolHandler(/ws, null, true, 64 * 1024, true, true, 10000L));pipeline.addLast(handlerWebSocket);}});//启动ChannelFuture channelFuture serverBootstrap.bind(appConfig.getWsPort()).sync();logger.info(Netty服务端启动成功,端口:{}, appConfig.getWsPort());channelFuture.channel().closeFuture().sync();} catch (Exception e) {e.printStackTrace();bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();} finally {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}
}创建并配置 ServerBootstrap ServerBootstrap 是 Netty 用于引导服务器的启动助手类。serverBootstrap.group(bossGroup, workerGroup) 设置了用于处理连接的 bossGroup 和用于处理消息的 workerGroup。serverBootstrap.channel(NioServerSocketChannel.class) 设置了服务器通道类型为 NioServerSocketChannel它适用于 NIO 传输。 添加处理器 serverBootstrap.handler(new LoggingHandler(LogLevel.DEBUG))添加一个日志处理器用于记录调试级别的日志。serverBootstrap.childHandler(new ChannelInitializerChannel() {...})添加一个子处理器用于初始化每个新连接的通道。 初始化通道 HttpServerCodec添加 HTTP 编解码器支持 HTTP 协议。HttpObjectAggregator添加 HTTP 对象聚合器确保接收完整的 HTTP 请求。IdleStateHandler添加空闲状态处理器用于检测读超时60秒。HandlerHeartBeat添加心跳检测处理器用于处理空闲事件。WebSocketServerProtocolHandler添加 WebSocket 协议处理器将 HTTP 协议升级为 WebSocket 协议。handlerWebSocket添加自定义的 WebSocket 消息处理器。 启动服务器 ChannelFuture channelFuture serverBootstrap.bind(appConfig.getWsPort()).sync();绑定端口并启动服务器。logger.info(Netty服务端启动成功,端口:{}, appConfig.getWsPort());记录服务器启动成功的日志。channelFuture.channel().closeFuture().sync();等待服务器通道关闭。 异常处理和资源释放 如果发生异常记录堆栈跟踪并优雅地关闭 bossGroup 和 workerGroup释放资源。
2.总结
NettyWebSocketStarter 是一个用于启动 Netty WebSocket 服务器的类。它通过配置一系列处理器如 HTTP 编解码器、心跳检测处理器、WebSocket 协议处理器等来初始化服务器并在指定端口上启动服务器。同时通过 PreDestroy 注解确保在 Spring 容器销毁时优雅地关闭资源。该类实现了 Runnable 接口使其可以在独立线程中运行通常可以用于多线程环境中启动 Netty 服务器。
3.Netty WebSocket 处理器 HandlerWebSocket
/*** 设置通道共享*/
ChannelHandler.Sharable
Component(handlerWebSocket)
public class HandlerWebSocket extends SimpleChannelInboundHandlerTextWebSocketFrame {private static final Logger logger LoggerFactory.getLogger(HandlerWebSocket.class);// Resource
// private ChannelContextUtils channelContextUtils;Resourceprivate RedisComponet redisComponet;/*** 当通道就绪后会调用此方法通常我们会在这里做一些初始化操作** param ctx* throws Exception*/Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {// Channel channel ctx.channel();logger.info(有新的连接加入。。。);}/*** 当通道不再活跃时连接关闭会调用此方法我们可以在这里做一些清理工作** param ctx* throws Exception*/Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {logger.info(有连接已经断开。。。);//channelContextUtils.removeContext(ctx.channel());}/*** 读就绪事件 当有消息可读时会调用此方法我们可以在这里读取消息并处理。** param ctx* param textWebSocketFrame* throws Exception*/Overrideprotected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame textWebSocketFrame) throws Exception {//接收心跳Channel channel ctx.channel();logger.info(接收到消息,{}, textWebSocketFrame.text());// AttributeString attribute channel.attr(AttributeKey.valueOf(channel.id().toString()));//String userId attribute.get();//redisComponet.saveUserHeartBeat(userId);}//用于处理用户自定义的事件 当有用户事件触发时会调用此方法例如连接超时异常等。Overridepublic void userEventTriggered(ChannelHandlerContext ctx, Object evt) {if (evt instanceof WebSocketServerProtocolHandler.HandshakeComplete) {WebSocketServerProtocolHandler.HandshakeComplete complete (WebSocketServerProtocolHandler.HandshakeComplete) evt;String url complete.requestUri();String token getToken(url);if (token null) {ctx.channel().close();return;}TokenUserInfoDto tokenUserInfoDto redisComponet.getTokenUserInfoDto(token);if (null tokenUserInfoDto) {ctx.channel().close();return;}/*** 用户加入*///channelContextUtils.addContext(tokenUserInfoDto.getUserId(), ctx.channel());}}/*** 获取url中的token** param url* return*/private String getToken(String url) {if (StringTools.isEmpty(url) || url.indexOf(?) -1) {return null;}String[] queryParams url.split(\\?);if (queryParams.length 2) {return url;}String[] params queryParams[1].split();if (params.length ! 2) {return url;}return params[1];}
}代码解析
这段代码定义了一个 Netty WebSocket 处理器 HandlerWebSocket继承自 SimpleChannelInboundHandlerTextWebSocketFrame。该处理器主要用于处理 WebSocket 的各种事件如连接建立、消息接收、连接关闭等。以下是对代码的详细解析
类和成员变量
ChannelHandler.Sharable
Component(handlerWebSocket)
public class HandlerWebSocket extends SimpleChannelInboundHandlerTextWebSocketFrame {private static final Logger logger LoggerFactory.getLogger(HandlerWebSocket.class);Resourceprivate RedisComponet redisComponet;
}ChannelHandler.Sharable注解表明这个处理器是可以在多个 Channel 之间共享的。Component(handlerWebSocket)将这个类注册为 Spring 的一个组件并指定组件名称为 handlerWebSocket。logger用于记录日志的 Logger 对象。redisComponet注入的 Redis 组件用于与 Redis 进行交互。
channelActive 方法
Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {logger.info(有新的连接加入。。。);
}channelActive当通道就绪时调用这个方法通常用于初始化操作。logger.info记录有新的连接加入的日志。
channelInactive 方法
Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {logger.info(有连接已经断开。。。);//channelContextUtils.removeContext(ctx.channel());
}channelInactive当通道不再活跃连接关闭时调用这个方法通常用于清理工作。logger.info记录有连接断开的日志。
channelRead0 方法
Override
protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame textWebSocketFrame) throws Exception {Channel channel ctx.channel();logger.info(接收到消息,{}, textWebSocketFrame.text());// AttributeString attribute channel.attr(AttributeKey.valueOf(channel.id().toString()));// String userId attribute.get();// redisComponet.saveUserHeartBeat(userId);
}channelRead0当有消息可读时调用这个方法读取并处理消息。Channel获取当前的通道。logger.info记录接收到的消息。注释掉的部分代码可以用于获取用户 ID 并保存心跳到 Redis。
userEventTriggered 方法
Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {if (evt instanceof WebSocketServerProtocolHandler.HandshakeComplete) {WebSocketServerProtocolHandler.HandshakeComplete complete (WebSocketServerProtocolHandler.HandshakeComplete) evt;String url complete.requestUri();String token getToken(url);if (token null) {ctx.channel().close();return;}TokenUserInfoDto tokenUserInfoDto redisComponet.getTokenUserInfoDto(token);if (null tokenUserInfoDto) {ctx.channel().close();return;}// 用户加入// channelContextUtils.addContext(tokenUserInfoDto.getUserId(), ctx.channel());}
}userEventTriggered处理用户自定义事件如连接超时、异常等。WebSocketServerProtocolHandler.HandshakeComplete处理 WebSocket 握手完成事件。getToken从 URL 中获取 token。ctx.channel().close()关闭通道如果 token 为空或无效。
getToken 方法
private String getToken(String url) {if (StringTools.isEmpty(url) || url.indexOf(?) -1) {return null;}String[] queryParams url.split(\\?);if (queryParams.length 2) {return url;}String[] params queryParams[1].split();if (params.length ! 2) {return url;}return params[1];
}getToken从 URL 中提取 token 参数。StringTools.isEmpty(url)检查 URL 是否为空。url.indexOf(?)检查 URL 是否包含查询参数。url.split(\\?)分割 URL 获取查询参数部分。params.length检查参数长度是否符合预期。
总结
HandlerWebSocket 是一个 Netty WebSocket 处理器用于处理各种 WebSocket 事件。它包含以下功能
channelActive当通道就绪时记录新的连接加入。channelInactive当通道不再活跃时记录连接断开。channelRead0读取并处理接收到的 WebSocket 消息记录消息内容。userEventTriggered处理 WebSocket 握手完成事件从 URL 中提取 token 并验证如果无效则关闭连接。getToken从 URL 中提取 token 参数用于验证用户身份。
通过这些方法HandlerWebSocket 能够有效地管理 WebSocket 连接的生命周期和消息处理确保连接的可靠性和安全性。 完整过程
当然可以我会详细解释从访问 WebSocket 接口到完成所有步骤的整个流程包括各个代码片段是如何协同工作的。
1. 客户端连接到 WebSocket 服务器
当客户端尝试连接到 WebSocket 服务器时例如通过 ws://localhost:5051/ws?token16515151Netty 会接受这个连接请求并触发一系列的事件来处理这个连接。
2. 服务器端初始化和配置
在服务器端我们使用了 NettyWebSocketStarter 类来初始化和配置 Netty 服务器。
Netty 服务器启动
Component
public class NettyWebSocketStarter implements Runnable {private static final Logger logger LoggerFactory.getLogger(NettyWebSocketStarter.class);Resourceprivate AppConfig appConfig;Resourceprivate HandlerWebSocket handlerWebSocket;private EventLoopGroup bossGroup new NioEventLoopGroup(1);private EventLoopGroup workerGroup new NioEventLoopGroup();PreDestroypublic void close() {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}Overridepublic void run() {try {ServerBootstrap serverBootstrap new ServerBootstrap();serverBootstrap.group(bossGroup, workerGroup);serverBootstrap.channel(NioServerSocketChannel.class).handler(new LoggingHandler(LogLevel.DEBUG)).childHandler(new ChannelInitializerChannel() {Overrideprotected void initChannel(Channel channel) {ChannelPipeline pipeline channel.pipeline();pipeline.addLast(new HttpServerCodec());pipeline.addLast(new HttpObjectAggregator(64 * 1024));pipeline.addLast(new IdleStateHandler(60, 0, 0, TimeUnit.SECONDS));pipeline.addLast(new HandlerHeartBeat());pipeline.addLast(new WebSocketServerProtocolHandler(/ws, null, true, 64 * 1024, true, true, 10000L));pipeline.addLast(handlerWebSocket);}});ChannelFuture channelFuture serverBootstrap.bind(appConfig.getWsPort()).sync();logger.info(Netty服务端启动成功,端口:{}, appConfig.getWsPort());channelFuture.channel().closeFuture().sync();} catch (Exception e) {e.printStackTrace();bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();} finally {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}
}3. 通道初始化
当客户端连接到服务器时Netty 会初始化通道并调用配置的处理器。这里的处理器包括 HttpServerCodec、HttpObjectAggregator、IdleStateHandler、HandlerHeartBeat 和 WebSocketServerProtocolHandler 以及我们的 HandlerWebSocket。
4. 握手和连接事件
WebSocket 协议处理和心跳检测
WebSocketServerProtocolHandler 处理 WebSocket 协议的握手。当握手完成时会触发 userEventTriggered 方法。
Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {if (evt instanceof WebSocketServerProtocolHandler.HandshakeComplete) {WebSocketServerProtocolHandler.HandshakeComplete complete (WebSocketServerProtocolHandler.HandshakeComplete) evt;String url complete.requestUri();String token getToken(url);if (token null) {ctx.channel().close();return;}TokenUserInfoDto tokenUserInfoDto redisComponet.getTokenUserInfoDto(token);if (null tokenUserInfoDto) {ctx.channel().close();return;}// 用户加入// channelContextUtils.addContext(tokenUserInfoDto.getUserId(), ctx.channel());}
}握手完成当 WebSocket 握手完成时WebSocketServerProtocolHandler 触发 HandshakeComplete 事件。提取 Token从 URL 中提取 token。验证 Token从 Redis 中验证 Token 是否有效。管理连接将用户信息和通道关联注释部分。
心跳检测
IdleStateHandler 会检测连接的空闲状态例如60秒内没有读取到数据触发相应的事件。
Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {if (evt instanceof IdleStateEvent) {IdleStateEvent e (IdleStateEvent) evt;if (e.state() IdleState.READER_IDLE) {AttributeString attribute ctx.channel().attr(AttributeKey.valueOf(ctx.channel().id().toString()));String userId attribute.get();logger.info(用户{}没有发送心跳断开连接, userId);ctx.close();} else if (e.state() IdleState.WRITER_IDLE) {ctx.writeAndFlush(heart);}}
}读空闲如果长时间未读取到数据READER_IDLE则关闭连接。写空闲如果长时间未写入数据WRITER_IDLE发送心跳消息。
5. 消息处理
当客户端发送消息时Netty 会触发 channelRead0 方法处理接收到的 WebSocket 消息。
Override
protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame textWebSocketFrame) throws Exception {Channel channel ctx.channel();logger.info(接收到消息,{}, textWebSocketFrame.text());// AttributeString attribute channel.attr(AttributeKey.valueOf(channel.id().toString()));// String userId attribute.get();// redisComponet.saveUserHeartBeat(userId);
}接收消息记录接收到的消息内容。心跳更新注释部分可以从消息中提取用户ID并在 Redis 中更新心跳记录。
6. 连接关闭
当连接关闭时Netty 会调用 channelInactive 方法。
Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {logger.info(有连接已经断开。。。);//channelContextUtils.removeContext(ctx.channel());
}记录日志记录连接断开的日志。清理资源注释部分从上下文中移除断开的连接。
总结
整个过程包括以下步骤
客户端连接客户端通过 WebSocket URL 连接到服务器如 ws://localhost:5051/ws?token16515151。服务器初始化NettyWebSocketStarter 启动服务器并配置各个处理器。握手和协议升级WebSocketServerProtocolHandler 处理握手并升级协议。事件触发和处理 握手完成在 userEventTriggered 中处理握手完成事件提取并验证 token。心跳检测在 HandlerHeartBeat 中处理读空闲和写空闲事件。 消息处理在 HandlerWebSocket 的 channelRead0 方法中处理接收到的 WebSocket 消息。连接关闭在 HandlerWebSocket 的 channelInactive 方法中处理连接关闭事件。
通过这些步骤Netty 能够高效地管理 WebSocket 连接确保连接的可靠性和数据的实时性。 文章转载自： http://www.morning.nba1on1.com.gov.cn.nba1on1.com http://www.morning.dnqpq.cn.gov.cn.dnqpq.cn http://www.morning.bfycr.cn.gov.cn.bfycr.cn http://www.morning.lkfsk.cn.gov.cn.lkfsk.cn http://www.morning.jfcbs.cn.gov.cn.jfcbs.cn http://www.morning.mbbgk.com.gov.cn.mbbgk.com http://www.morning.kxsnp.cn.gov.cn.kxsnp.cn http://www.morning.ktfbl.cn.gov.cn.ktfbl.cn http://www.morning.nslwj.cn.gov.cn.nslwj.cn http://www.morning.nfgbf.cn.gov.cn.nfgbf.cn http://www.morning.hrpbq.cn.gov.cn.hrpbq.cn http://www.morning.splcc.cn.gov.cn.splcc.cn http://www.morning.nlkm.cn.gov.cn.nlkm.cn http://www.morning.fjmfq.cn.gov.cn.fjmfq.cn http://www.morning.txtzr.cn.gov.cn.txtzr.cn http://www.morning.wqmyh.cn.gov.cn.wqmyh.cn http://www.morning.qfnrx.cn.gov.cn.qfnrx.cn http://www.morning.gxcit.com.gov.cn.gxcit.com http://www.morning.mxhcf.cn.gov.cn.mxhcf.cn http://www.morning.hqllx.cn.gov.cn.hqllx.cn http://www.morning.yfmwg.cn.gov.cn.yfmwg.cn http://www.morning.fllfc.cn.gov.cn.fllfc.cn http://www.morning.gpryk.cn.gov.cn.gpryk.cn http://www.morning.lmcrc.cn.gov.cn.lmcrc.cn http://www.morning.ldqrd.cn.gov.cn.ldqrd.cn http://www.morning.ktmbp.cn.gov.cn.ktmbp.cn http://www.morning.tbnpn.cn.gov.cn.tbnpn.cn http://www.morning.bykqg.cn.gov.cn.bykqg.cn http://www.morning.rqhbt.cn.gov.cn.rqhbt.cn http://www.morning.sgwr.cn.gov.cn.sgwr.cn http://www.morning.rxpp.cn.gov.cn.rxpp.cn http://www.morning.kdgcx.cn.gov.cn.kdgcx.cn http://www.morning.rdmn.cn.gov.cn.rdmn.cn http://www.morning.rnsjp.cn.gov.cn.rnsjp.cn http://www.morning.yntsr.cn.gov.cn.yntsr.cn http://www.morning.gqryh.cn.gov.cn.gqryh.cn http://www.morning.npqps.cn.gov.cn.npqps.cn http://www.morning.rxnxl.cn.gov.cn.rxnxl.cn http://www.morning.lwcqh.cn.gov.cn.lwcqh.cn http://www.morning.sxjmz.cn.gov.cn.sxjmz.cn http://www.morning.mdtfh.cn.gov.cn.mdtfh.cn http://www.morning.mhybs.cn.gov.cn.mhybs.cn http://www.morning.nbnq.cn.gov.cn.nbnq.cn http://www.morning.hgwsj.cn.gov.cn.hgwsj.cn http://www.morning.rythy.cn.gov.cn.rythy.cn http://www.morning.sltfk.cn.gov.cn.sltfk.cn http://www.morning.kcypc.cn.gov.cn.kcypc.cn http://www.morning.qbgdy.cn.gov.cn.qbgdy.cn http://www.morning.splcc.cn.gov.cn.splcc.cn http://www.morning.mbprq.cn.gov.cn.mbprq.cn http://www.morning.ctwwq.cn.gov.cn.ctwwq.cn http://www.morning.hwlk.cn.gov.cn.hwlk.cn http://www.morning.pthmn.cn.gov.cn.pthmn.cn http://www.morning.znrgq.cn.gov.cn.znrgq.cn http://www.morning.hbqhz.cn.gov.cn.hbqhz.cn http://www.morning.tqbyw.cn.gov.cn.tqbyw.cn http://www.morning.rahllp.com.gov.cn.rahllp.com http://www.morning.gktds.cn.gov.cn.gktds.cn http://www.morning.kuaijili.cn.gov.cn.kuaijili.cn http://www.morning.qqrqb.cn.gov.cn.qqrqb.cn http://www.morning.dkslm.cn.gov.cn.dkslm.cn http://www.morning.tgtsg.cn.gov.cn.tgtsg.cn http://www.morning.jrlgz.cn.gov.cn.jrlgz.cn http://www.morning.kzslk.cn.gov.cn.kzslk.cn http://www.morning.lhrxq.cn.gov.cn.lhrxq.cn http://www.morning.nhgfz.cn.gov.cn.nhgfz.cn http://www.morning.hmhdn.cn.gov.cn.hmhdn.cn http://www.morning.pznhn.cn.gov.cn.pznhn.cn http://www.morning.mqmxg.cn.gov.cn.mqmxg.cn http://www.morning.xsszn.cn.gov.cn.xsszn.cn http://www.morning.xdnhw.cn.gov.cn.xdnhw.cn http://www.morning.zrrgx.cn.gov.cn.zrrgx.cn http://www.morning.gghhmi.cn.gov.cn.gghhmi.cn http://www.morning.mtmph.cn.gov.cn.mtmph.cn http://www.morning.ljngm.cn.gov.cn.ljngm.cn http://www.morning.gmmyn.cn.gov.cn.gmmyn.cn http://www.morning.mtsck.cn.gov.cn.mtsck.cn http://www.morning.bmgdl.cn.gov.cn.bmgdl.cn http://www.morning.rdlong.com.gov.cn.rdlong.com http://www.morning.qggcc.cn.gov.cn.qggcc.cn
