如何在不关闭连接的netty中设置读取超时？

Question

我有一个 netty 服务器，它接收来自客户端的请求，向另一台服务器发出请求，然后在对原始客户端请求的响应中使用来自第二个请求的响应。我希望对第二台服务器的请求有一个短暂的超时（约 40 毫秒），以便在超时时发送通用响应，但我不想关闭与第二台服务器的连接。相反，当超时请求到达时，我将简单地丢弃对超时请求的响应，然后将连接返回到我的池。

在 netty 中最好的方法是什么？我试过 ReadTimeoutHandler 但这似乎会在发生超时时关闭与第二台服务器的连接。

Answer 1

看来，根据您的要求，您需要围绕与第二台服务器的连接编写自己的时间管理。此外，由于您提到了一个连接池，这可能意味着您对可以同时与第二台服务器建立的连接数也有限制（嗯，总是有限制的，这取决于您是否真的需要担心它与否）。

对于您不担心排队出站请求的简单情况，您应该能够创建 ReadTimeoutHandler 的副本并根据需要对其进行修改，最有可能的是让它从上下文中拉出一个回调以调用它而不是在其 readTimedOut() 方法中关闭连接。

如果您要将出站请求排队，您应该考虑在队列上花费的时间以及接收响应的时间，因此您的解决方案需要您自己的计时器，该计时器在您放置后立即启动一个项目进入出站队列。此外，您需要一种方法在计时器和返回的有效结果之间进行同步（您只想发送一个响应，而不是在计时器超时时发送一个响应，以及在来自第二个服务器的响应到来时发送一个响应）。为此，我想您可能希望使用某种类型的管理器对象，该对象包含对通道的引用，并且您的回调通过该管理器回写。

例如（在伪代码中）

MyStateManager manager = new MyStateManager(channel);
// Scheduled a timer task for 40ms from now to invoke the callback
// method of your ReadTimedOutCallback and send a message through the manager
myTimer.schedule(new ReadTimedOutCallback(manager),40);
// Send an outbound request to server 2 through a class that queues
// and manages such requests asynchronously.  When the request is complete
// it invokes the callback method of RequestCompletedCallback which then
// sends a response through the manager.
myAsyncOutboundServerRequester.request(new RequestCompletedCallback(manager));
// ... finish and return the thread to Netty for use with other requests
// while the async tasks process

简单形式的管理器类似于（不包括异常处理、通道状态检查等）：

public class MyStateManager
{
    private final Channel channel;
    private AtomicBoolean messageSent = new AtomicBoolean(false);
    public MyStateManager(Channel channel)
    {
        this.channel = channel;
    }

    // Called by the ReadTimeoutCallback
    public void sendGenericResponse()
    {
        if (messageSent.getAndSet(true))
        {
           //... write generic response to channel
           ChannelFuture future = channel.write... 
           // Add listeners to future, etc
        } 
    }

    // Called by the RequestCompletedCallback
    public void sendResponse(MyResponseObject response)
    {
        if (messageSent.getAndSet(true))
        {
            // write returned response to channel
            ChannelFuture future = channel.write(response);
            // Add listeners to future, etc
        }
    }

}

您可能仍希望对与第二台服务器的连接进行某种类型的超时检查，以便在连接太长时间（30 秒或一分钟或其他时间）无响应时关闭连接。