Netty的超时机制

原文地址：http://blog.163.com/linfenliang@126/blog/static/127857195201467112958326/


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。