Netty Comet异步请求超时

本文介绍了Netty Comet异步请求超时的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！

问题描述

我尝试使用Jboss Netty创建长时间轮询Comet。

如何在30秒内配置时间？以下文档：

  @Override 
 public ChannelPipeline getPipeline（）throws Exception {
 ChannelPipeline pipeline = Channels 。管道（）; 
 pipeline.addLast（decoder，new HttpRequestDecoder（））; 
 pipeline.addLast（encoder，new HttpResponseEncoder（））; 
 pipeline.addLast（handler，new HTTPRequestHandler（））; 
 timer timer = new HashedWheelTimer（）; 
 pipeline.addLast（timeout，new IdleStateHandler（timer，30，30，0））; 
 return pipeline;

但它不工作，我的请求永远持续。

这是否意味着我需要实现 Callable< T> ，然后调用 Future.get 使用超时参数和终止请求如果 TimeOutException 发生？那么我应该使用 Future< ChannelFuture> 吗？

有没有其他方法？

代码：

  FutureExecutor executor = FutureExecutor.getInstance（）; 
 
 @Override 
 public void messageReceived（ChannelHandlerContext ctx，MessageEvent e）throws Exception {
 HttpRequest request =（HttpRequest）e.getMessage（）; 
 ChannelFuture channelFuture = null; 
 Callable< ChannelFuture> myRunnable = new MyCallable（e）; 
 Future< ChannelFuture> future = executor.fireEvent（myRunnable）; 
 
 try {
 channelFuture = future.get（40，TimeUnit.SECONDS）; 
} catch（TimeoutException ex）{
 
 channelFuture = e.getChannel（Response timeOutResponse）; 
 //处理超时
} catch（InterruptedException ex）{
 channelFuture = e.getChannel（Response interaptedResponse）; 
 
} catch（ExecutionException ex）{
 channelFuture = e.getChannel（Response errorResponse）; 
} 
 finally {
 future.cancel（true）; 
 channelFuture.addListener（ChannelFutureListener.CLOSE）; 
} 
 
}

只监视BlockedQueue：

  @Override 
 public ChannelFuture call（）{
 final BlockingQueue& queue = ..... 
 while（true）{
消息消息= queue.take（）; 
 ChannelBuffer partialresponse = ChannelBuffers.buffer（message.toJson（））; 
 ChannelFuture future = e.getChannel（）。write（partialresponse）; 
 return future; 
} 
}

解决方案

的你应该在你的管道之间共享一个HashedWheelTimer的实例，因为它将为每个实例创建一个线程。但现在你的问题..

如果你使用IdleStateHandler，你还需要实现一个IdleStateAwareHandler或IdleStateAwareChannelUpstreamHandler，它将对由IdleStateHandler触发的IdleState事件做出反应。所以，如果你想要例如断开频道后，它是空闲的，你可以只是调用Channel.close（）之后接收到该事件。

另请参阅： / p>

另一个解决方案是添加ReadTimeoutHandler，然后当您在exceptionCaught（..）方法中捕获它时执行ReadTimeoutException。

参见：

I'm trying to create long polling Comet using Jboss Netty.

How can I configure time out of 30 sec? Following the documentaton:

@Override
    public ChannelPipeline getPipeline() throws Exception {
    ChannelPipeline pipeline = Channels.pipeline();
     pipeline.addLast("decoder", new HttpRequestDecoder());
     pipeline.addLast("encoder", new HttpResponseEncoder());
     pipeline.addLast("handler", new HTTPRequestHandler());
     Timer timer = new HashedWheelTimer();
     pipeline.addLast("timeout", new IdleStateHandler(timer, 30, 30, 0));
    return pipeline;

but it doesn't work and my request lasts forever. How can this be solved?

Does it mean that I need to implement Callable<T>, then call Future.get with timeout parameter and terminate request if TimeOutException occurs? Then should I use Future<ChannelFuture>?

Is there any other approach?

Code:

FutureExecutor executor = FutureExecutor.getInstance();

    @Override
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
        HttpRequest request = (HttpRequest) e.getMessage();
    ChannelFuture channelFuture = null;
        Callable<ChannelFuture> myRunnable = new MyCallable(e);
        Future<ChannelFuture> future = executor.fireEvent(myRunnable);

    try{
        channelFuture = future.get(40,TimeUnit.SECONDS);
        }catch (TimeoutException ex) {

        channelFuture = e.getChannel(Response timeOutResponse);
               // handle the timeout
        } catch (InterruptedException ex) {
        channelFuture = e.getChannel(Response interaptedResponse);

        } catch (ExecutionException ex) {
            channelFuture = e.getChannel(Response errorResponse);
        }
        finally{
            future.cancel(true);
            channelFuture.addListener(ChannelFutureListener.CLOSE);
        }

}

and inside of Callable I'm just monitoring BlockedQueue:

@Override
public ChannelFuture call() {
        final BlockingQueue<String> queue  =.....
        while (true){
        Message message = queue.take();
                ChannelBuffer partialresponse = ChannelBuffers.buffer(message.toJson());
         ChannelFuture future = e.getChannel().write(partialresponse);
        return future;
        }
}

解决方案

First of you should share one instance of HashedWheelTimer between your pipelines, as it will create one thread per instance. But now to your problem..

If you use a IdleStateHandler you also need to implement an IdleStateAwareHandler or IdleStateAwareChannelUpstreamHandler which will react on the IdleState events which get triggered by the IdleStateHandler. So if you want for example to disconnect the Channel after it was idle for you could just call Channel.close() on it after receiving the event.