在 DotNetty 中实现同步请求

一、背景

DotNetty 本身是一个优秀的网络通讯框架，不过它是基于异步事件驱动来处理另一端的响应，需要在单独的 Handler 去处理相应的返回结果。而在我们的实际使用当中，尤其是 客户端程序 基本都是 请求-响应 模型，在发送了数据时候需要等待服务器的响应才能进行下一步操作，如果服务器返回的是错误信息，则需要进行特殊的处理。

类似于下面这种方式：

public async void Button1_Click()
{
var result = await DotNettyClient.SendData("Hello");

if(result == "Error")
    {
throw new Exception("服务器返回错误!");
    }

    Console.WriteLine($"Hello {result}");
}

二、解决思路

参阅了大部分资料之后，发现在 Java 的 Netty 当中可以使用 Future / Promise 来实现，那么 C# 是否有类似的组件呢？答案是有的，他们对应的就是 Task 和 TaskCompletionSource，前者是给调用者的任务，而后者则是用于设置响应任务的结果。

那么我们就可以这么来处理，当客户端发送请求时，附带唯一的一个请求 ID，并将 TaskCompletionSource 放在一个请求队列当中，之后返回一个 Task。当客户端接收到服务器响应的时候，通过 TaskCompletionSource 设置之前那个 Task 的结果，这样我们接收到响应之后，就会从之前 await 的地方继续执行。

这里我自己的需求仅仅是类似于 同步阻塞式 的操作，所以我直接使用一个队列来处理，并没有用唯一的请求 ID 来表示不同的请求，因为我可以 保证在同一时间内有且仅有一个客户端请求被发起。

三、代码实现

实现起来超级简单，只需要在发起请求的时候，创建一个 TaskCompletionSource<TResponse> 对象。这个泛型参数指的是你想要的返回值类型，这里我以 TResponse 代替，下面的 DEMO 我会用 string 类型进行演示。

创建好一个 TaskCompletionSource<TResponse> 之后，在发送方法里面，我们可以将其对象放在一个先进先出的队列当中，然后将其 Task 属性作为发送方法的返回值。

我们再来到处理服务器响应的 Handler 当中，从队列里面拿去之前存放的 TaskCompletionSource<TResponse> 对象，调用其 SetResult() 方法，将具体响应进行设置。

通过以上的操作，我们在发送数据的时候，就可以使用 await 关键字等待服务端的响应，但不会阻塞线程，当客户端接收到服务端响应时，就会恢复到之前 await 的位置继续执行。

数据发送方法：

public static class DotNettyClient
{
static DotNettyClient()
{
        RequestQueue = new Queue<TaskCompletionSource<string>>();
    }

public static Queue<TaskCompletionSource<string>> RequestQueue { get; set; }

public static async Task<string> SendData(string data)
{
var resultTask = new TaskCompletionSource<string>();

var buffer = new Unpooled.Buffer();
        buffer.WriteBytes(Encoding.UTF8.GetBytes(data));
await _clientChannel.WriteAndFlushAsync(buffer);

        RequestQueue.Enqueue(resultTask);

return await resultTask.Task;
    }
}

服务端响应处理：

public class ProtocolHandler : ChannelHandlerAdapter
{
public override void ChannelRead(IChannelHandlerContext context, object message)
{
if(message is string response)
        {
if(!DotNettyClient.RequestQueue.TryDequeue(out TaskCompletionSource<string> result)) return;
            result.SetResult(response);
        }
    }
}