Netty超时机制学习
-
技术点描述
- ReadTimeoutHandler读取数据超时处理
- WriteTimeoutHandler写数据超时处理
-
IdleStateHandler状态空闲处理
通过以上三种方式,可以轻松实现SOCKET长连接的心跳包机制。
另要注意的是这里所说的超时是指逻辑上的超时,而非TCP连接超时。
-
实现方案
ReadTimeoutHandler,WriteTimeoutHandler, IdleStateHandler这三种方式都是通过在数据处理管道ChannelPipelineFactory的实现类里配置相关代码来实现。
一般使用根据需求使用一种方式即可。
它主要使用Timer定时器用自定义的时间,去定时检测相应的状态,检测到后会根据选择的处理方式,是报错或回调指定的接口实现类处理。
-
参考源码包
-
主要包
-
IdleStateHandler 空闲状态的处理代码
源码:
/**
* Creates a new instance.
*
* @param timer
* the Timer that is used to trigger the scheduled event.
* The recommended Timer implementation is HashedWheelTimer.
* @param readerIdleTimeSeconds
* an IdleStateEvent whose state is IdleState.READER_IDLE
* will be triggered when no read was performed for the specified
* period of time. Specify 0 to disable.
* @param writerIdleTimeSeconds
* an IdleStateEvent whose state is IdleState.WRITER_IDLE
* will be triggered when no write was performed for the specified
* period of time. Specify {@code 0} to disable.
* @param allIdleTimeSeconds
* an IdleStateEvent whose state is IdleState.ALL_IDLE
* will be triggered when neither read nor write was performed
* for the specified period of time. Specify 0 to disable.
*/
public IdleStateHandler(
Timer timer,
int readerIdleTimeSeconds,
int writerIdleTimeSeconds,
int allIdleTimeSeconds) {
this(timer, readerIdleTimeSeconds, writerIdleTimeSeconds, allIdleTimeSeconds, TimeUnit.SECONDS);
}
-
ReadTimeoutHandler 读取数据超时的处理代码
源码:
/**
* Creates a new instance.
*
* @param timer
* the Timer that is used to trigger the scheduled event.
* The recommended Timer implementation is HashedWheelTimer.
* @param timeoutSeconds
* read timeout in seconds
*/
public ReadTimeoutHandler(Timer timer, int timeoutSeconds) {
this(timer, timeoutSeconds, TimeUnit.SECONDS);
}
-
WriteTimeoutHandler 写数据超时的处理代码
源码:
/**
* Creates a new instance.
*
* @param timer
* the Timer that is used to trigger the scheduled event.
* The recommended Timer implementation is HashedWheelTimer.
* @param timeoutSeconds
* write timeout in seconds
*/
public WriteTimeoutHandler(Timer timer, int timeoutSeconds) {
this(timer, timeoutSeconds, TimeUnit.SECONDS);
}
-
辅助包
-
IdleStateEvent 空闲事件的接口类
源码:
/**
* A ChannelEvent that is triggered when a Channel has been idle
* for a while.
* @apiviz.landmark
* @apiviz.has org.jboss.netty.handler.timeout.IdleState oneway - -
*/
public interface IdleStateEvent extends ChannelEvent {
/**
* Returns the detailed idle state.
*/
IdleState getState();
/**
* Returns the last time when I/O occurred in milliseconds.
*/
long getLastActivityTimeMillis();
}
- DefaultIdleStateEvent空闲事件的实现类
-
IdleStateAwareChannelHandler 处理空闲事件的接口.
源码:
/**
* An extended SimpleChannelHandler that adds the handler method for
* an IdleStateEvent.
* @apiviz.uses org.jboss.netty.handler.timeout.IdleStateEvent
*/
public class IdleStateAwareChannelHandler extends SimpleChannelHandler {
/**
* Creates a new instance.
*/
public IdleStateAwareChannelHandler() {
super();
}
@Override
public void handleUpstream(ChannelHandlerContext ctx, ChannelEvent e) throws Exception {
if (e instanceof IdleStateEvent) {
channelIdle(ctx, (IdleStateEvent) e);
} else {
super.handleUpstream(ctx, e);
}
}
/**
* Invoked when a Channel has been idle for a while.
*/
public void channelIdle(ChannelHandlerContext ctx, IdleStateEvent e) throws Exception {
ctx.sendUpstream(e);
}
}
- IdleStateAwareChannelUpstreamHandler处理空闲事件的接口.
-
IdleState 定义了三种空闲状态的枚举类
源码:
/**
* An Enum that represents the idle state of a Channel.
*/
public enum IdleState {
/**
* No data was received for a while.
*/
READER_IDLE,
/**
* No data was sent for a while.
*/
WRITER_IDLE,
/**
* No data was either received or sent for a while.
*/
ALL_IDLE;
}
- ReadTimeoutException 读取超时异常类
- WriteTimeoutException 写数据超时异常类
- TimeoutException 读取数据和写数据异常类的父类
-
Demo实现
以下是关于三种方式的关键代码实现讲解。
-
IdleStateHandler
//第一步,在服务启动类中设置数据处理管道.
...
bootstrap.setPipelineFactory(new TCPServerPipelineFactory());
...
//第二步,在数据处理管道实现类里配置空闲状态处理代码.
public class TCPServerPipelineFactory implements ChannelPipelineFactory {
@Override
public ChannelPipeline getPipeline() throws Exception {
// Create a default pipeline implementation.
ChannelPipeline pipeline = Channels.pipeline();
//设置空闲状态处理操作
pipeline.addLast("idlehandler", new IdleStateHandler(new HashedWheelTimer(), 10, 5 , 0));
pipeline.addLast("hearbeat", new Heartbeat());
pipeline.addLast("handler", new TCPServerHandler());
return pipeline;
}
//第三步,实现方式1:自定义IdleStateAwareChannelHandler的实现类,
//当Channel处理空闲状态时,会触发此方法.
public class Heartbeat extends IdleStateAwareChannelHandler {
@Override
public void channelIdle(ChannelHandlerContext ctx, IdleStateEvent e) throws Exception {
super.channelIdle(ctx, e);
if (e.getState() == IdleState.READER_IDLE){
byte[] test = " ac0bce0490007050006".getBytes();
ChannelBuffer channelBuffer = ChannelBuffers.buffer(test.length);
channelBuffer.writeBytes(upData);
//发送超时数据到终端.
e.getChannel().write(channelBuffer);
}
}
//第三步, 实现方式2:在扩展SimpleChannelHandler的handler类里,如下操作:
//记得注释掉第二步中的代码pipeline.addLast("hearbeat", new Heartbeat());
@Override
public void handleUpstream(ChannelHandlerContext ctx, ChannelEvent e) throws Exception {
if (e instanceof IdleStateEvent) {
IdleStateEvent ise = (IdleStateEvent)e;
if (ise.getState() == IdleState.READER_IDLE){
byte[] test = "超时 ...".getBytes();
ChannelBuffer channelBuffer = ChannelBuffers.buffer(test.length);
channelBuffer.writeBytes(test);
//发送超时数据到终端.
ctx.getChannel().write(channelBuffer);
}
}
super.handleUpstream(ctx, e);
}
-
ReadTimeoutHandler
//第一步,在服务启动类中设置数据处理管道.
...
bootstrap.setPipelineFactory(new TCPServerPipelineFactory());
...
//第二步,在数据处理管道实现类里配置空闲状态处理代码.
public class TCPServerPipelineFactory implements ChannelPipelineFactory {
@Override
public ChannelPipeline getPipeline() throws Exception {
// Create a default pipeline implementation.
ChannelPipeline pipeline = Channels.pipeline();
//设置读取数据超时处理
pipeline.addLast("readTimeOut",new ReadTimeoutHandler(new HashedWheelTimer(),10));
pipeline.addLast("handler", new TCPServerHandler());
return pipeline;
}
//第三步, 在扩展SimpleChannelHandler的handler类里,如下操作:
@Override
public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {
//对读取超时异常进行判断
if (e.getCause() instanceof ReadTimeoutException) {
byte[] test = "超时 ...".getBytes();
ChannelBuffer channelBuffer = ChannelBuffers.buffer(test.length);
channelBuffer.writeBytes(test);
ctx.getChannel().write(channelBuffer);
} else {
logger.log(Level.WARN, "Unexpected exception from downstream.",e.getCause());
}
}
-
WriteTimeoutHandler
略,原理同ReadTimeoutHandler。
相关推荐
配置maven添加io.netty 在com.zhao的包下的文件,可以自行修改使用
#### 心跳机制 - 服务端采用 `IdleStateHandler`,在一段时间内(默认15s)没有读到客户端消息则说明客户端已离线,服务器会触发读超时事件断开连接 - 客户端采用定时(默认10s)任务方式向服务端发送一个ping消息...
Netty的心跳检测和Keepalive机制都是为了确保客户端和服务器之间的连接仍然有效,防止连接断开。但它们在实现方式和原理上有所不同。Netty的心跳检测机制是一种自定义的、基于应用层的机制。它主要通过定时发送和...
使用方式1. 依赖im_lib库,implementation project(':im_lib')2. 自定义IMSEventListener,实现OnEventListener,重写对应的方法配置参数3. 自定义IMSConnectStatusListener,实现IMSConnectStatusCallback,重现...
基于Netty+TCP+Protobuf实现的Android IM库,包含Protobuf序列化、TCP拆包与粘包、长连接握手认证、心跳机制、断线重连机制、消息重发机制、读写超时机制、离线消息、线程池等功能
Hermes是一款基于Netty的可以支持百万级别的并发连接的高性能、高度可扩展的的网络通讯框架,它参考了dubbo和sofa-bolt的网络通讯模块的设计,hemers可以使用在IM、长连接等领域,它具有以下的特性: 私有的通讯...
基于Netty TCP Protobuf实现的Android IM库,包含Protobuf序列化、TCP拆包与粘包、长连接握手认证、心跳机制、断线重连机制、消息重发机制、读写超时机制、离线消息、线程池等功能。
| Netty | 高性能消息收发、心跳检测、应用层ACK | | Redis | 保存用户会话信息、去重信息、群聊单聊的离线消息以及其他信息的缓存 | | Mysql | 持久化信息 | | RabbitMQ | 将存储操作异步,利用RabbitMQ的可靠性...
( protocol-skeleton )命令与命令处理器编解码处理器心跳触发器3、私有协议定制实现 - RPC 通信协议 ( protocol-implementation )RPC 通信协议的设计灵活的反序列化时机控制请求处理超时 FailFast 机制用户请求...
SOFABolt 是蚂蚁金融服务集团开发的一套基于 Netty 实现的网络通信框架。 为了让 Java 程序员能将更多的精力放在基于网络通信的业务逻辑实现...请求处理超时 FailFast 机制 用户请求处理器 ( UserProcessor ) 双工通信