Thực hiện các tin nhắn lưu giữ trong Netty bằng cách sử dụng WriteTimeoutHandler

Question

Tôi đang sử dụng Netty 3.2.7. Tôi đang cố gắng viết chức năng trong ứng dụng khách của mình sao cho nếu không có thông báo nào được viết sau một khoảng thời gian nhất định (ví dụ, 30 giây), một thông báo "tiếp tục sống" được gửi đến máy chủ.

Sau một số lần đào, tôi thấy rằng WriteTimeoutHandler sẽ cho phép tôi thực hiện việc này. Tôi đã tìm thấy giải thích này tại đây: https://issues.jboss.org/browse/NETTY-79.

Ví dụ được đưa ra trong các tài liệu Netty là:

public ChannelPipeline getPipeline() { 
    // An example configuration that implements 30-second write timeout: 
    return Channels.pipeline(
     new WriteTimeoutHandler(timer, 30), // timer must be shared. 
     new MyHandler()); 
} 

Trong khách hàng thử nghiệm của tôi, tôi đã làm chỉ này. Trong myhandler, tôi cũng overrided phương pháp exceptionCaught():

public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) { 
    if (e.getCause() instanceof WriteTimeoutException) { 
     log.info("Client sending keep alive!"); 
     ChannelBuffer keepAlive = ChannelBuffers.buffer(KEEP_ALIVE_MSG_STR.length()); 
     keepAlive.writeBytes(KEEP_ALIVE_MSG_STR.getBytes()); 
     Channels.write(ctx, Channels.future(e.getChannel()), keepAlive); 
    } 
} 

Không có vấn đề gì thời gian khách hàng không viết bất cứ điều gì để các kênh, các exceptionCaught() phương pháp tôi đã ghi đè không bao giờ được gọi.

Nhìn vào nguồn gốc của WriteTimeoutHandler, writeRequested() thực hiện của nó là:

public void writeRequested(ChannelHandlerContext ctx, MessageEvent e) 
     throws Exception { 

    long timeoutMillis = getTimeoutMillis(e); 
    if (timeoutMillis > 0) { 
     // Set timeout only when getTimeoutMillis() returns a positive value. 
     ChannelFuture future = e.getFuture(); 
     final Timeout timeout = timer.newTimeout(
       new WriteTimeoutTask(ctx, future), 
       timeoutMillis, TimeUnit.MILLISECONDS); 

     future.addListener(new TimeoutCanceller(timeout)); 
    } 

    super.writeRequested(ctx, e); 
} 

Ở đây, có vẻ như rằng việc thực hiện này cho biết: "Khi một ghi được yêu cầu, làm một thời gian chờ mới Khi ghi thành công. , hủy thời gian chờ. "

Sử dụng trình gỡ lỗi, có vẻ như đây là những gì đang xảy ra. Ngay khi ghi xong, thời gian chờ sẽ bị hủy. Đây không phải là hành vi tôi muốn. Hành vi mà tôi muốn là: "Nếu khách hàng chưa viết bất kỳ thông tin nào cho kênh trong 30 giây, hãy ném một WriteTimeoutException".

Vì vậy, đây không phải là những gì WriteTimeoutHandler là dành cho? Đây là cách tôi giải thích nó từ những gì tôi đã đọc trực tuyến, nhưng việc thực hiện dường như không hoạt động theo cách này. Tôi có sử dụng sai không? Tôi có nên sử dụng cái gì khác không? Trong phiên bản Mina của cùng một máy khách tôi đang cố gắng viết lại, tôi thấy rằng phương thức sessionIdle() bị ghi đè để đạt được hành vi mà tôi muốn, nhưng phương thức này không có sẵn trong Netty.

Answer 1

Tôi sẽ đề xuất thêm IdleStateHandler và sau đó thêm triển khai tùy chỉnh của bạn IdleStateAwareUpstreamHandler có thể phản ứng ở trạng thái không hoạt động. Điều này làm việc rất tốt cho tôi trên nhiều dự án khác nhau.

Các javadocs liệt kê các ví dụ sau đây, mà bạn có thể sử dụng như là cơ sở thực hiện của bạn:

public class MyPipelineFactory implements ChannelPipelineFactory { 

    private final Timer timer; 
    private final ChannelHandler idleStateHandler; 

    public MyPipelineFactory(Timer timer) { 
     this.timer = timer; 
     this.idleStateHandler = new IdleStateHandler(timer, 60, 30, 0); 
     // timer must be shared. 
    } 

    public ChannelPipeline getPipeline() { 
     return Channels.pipeline(
      idleStateHandler, 
      new MyHandler()); 
    } 
} 

// Handler should handle the IdleStateEvent triggered by IdleStateHandler. 
public class MyHandler extends IdleStateAwareChannelHandler { 

    @Override 
    public void channelIdle(ChannelHandlerContext ctx, IdleStateEvent e) { 
     if (e.getState() == IdleState.READER_IDLE) { 
      e.getChannel().close(); 
     } else if (e.getState() == IdleState.WRITER_IDLE) { 
      e.getChannel().write(new PingMessage()); 
     } 
    } 
} 

ServerBootstrap bootstrap = ...; 
Timer timer = new HashedWheelTimer(); 
... 
bootstrap.setPipelineFactory(new MyPipelineFactory(timer)); 
... 

Answer 2

Đối Netty 4.0 và mới hơn, bạn nên mở rộng ChannelDuplexHandler như trong ví dụ từ IdleStateHandler documentation:

// An example that sends a ping message when there is no outbound traffic 
// for 30 seconds. The connection is closed when there is no inbound traffic 
// for 60 seconds. 

public class MyChannelInitializer extends ChannelInitializer<Channel> { 
    @Override 
    public void initChannel(Channel channel) { 
     channel.pipeline().addLast("idleStateHandler", new IdleStateHandler(60, 30, 0)); 
     channel.pipeline().addLast("myHandler", new MyHandler()); 
    } 
} 

// Handler should handle the IdleStateEvent triggered by IdleStateHandler. 
public class MyHandler extends ChannelDuplexHandler { 
    @Override 
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { 
     if (evt instanceof IdleStateEvent) { 
      IdleStateEvent e = (IdleStateEvent) evt; 
      if (e.state() == IdleState.READER_IDLE) { 
       ctx.close(); 
      } else if (e.state() == IdleState.WRITER_IDLE) { 
       ctx.writeAndFlush(new PingMessage()); 
      } 
     } 
    } 
}