当前位置：首页 > news >正文

【SuperSocket 】利用 TaskCompletionSource 在 SuperSocket 中实现跨模块异步处理客户端消息

news 2025/9/5 18:46:38

利用 TaskCompletionSource 在 SuperSocket 中实现跨模块异步处理客户端消息

在使用 SuperSocket 构建 TCP 服务时，我们经常会遇到这样的需求：

服务端接收到客户端数据后，需要将数据交给其他模块处理
处理完成后再将结果返回给调用模块或客户端
希望调用模块能够异步等待处理结果，而不是阻塞线程

本文将通过 TaskCompletionSource 来实现这一场景，并结合 SuperSocket 的异步回调机制，讲解完整实现方法。

1️⃣ TaskCompletionSource 基础语法

TaskCompletionSource<T>（简称 TCS）是 .NET 中用于手动控制 Task 完成时机的工具。

1.1 基本概念

Task：异步操作的表示，通常由方法内部执行完成
TaskCompletionSource：你可以手动控制 Task 何时完成，以及完成结果是什么

var tcs = new TaskCompletionSource<int>();
Task<int> task = tcs.Task;// 模拟异步事件
Task.Run(async () =>
{await Task.Delay(1000);   // 模拟耗时操作tcs.SetResult(42);        // 手动完成 Task 并返回结果
});int result = await task;
Console.WriteLine(result);    // 输出 42

核心思想：TaskCompletionSource 就像一个“空盒子”，你自己决定什么时候放入结果并“打开盒子”。

1.2 TCS 常用方法

方法	功能
`SetResult(T result)`	设置 Task 成功完成并返回结果
`TrySetResult(T result)`	安全写法，如果 Task 已完成不会抛异常
`SetException(Exception ex)`	设置 Task 异常完成
`SetCanceled()`	取消 Task

2️⃣ 在 SuperSocket 中使用 TCS

SuperSocket 的核心是事件驱动，客户端数据到达时会触发 UsePackageHandler 回调。我们可以利用 TCS，将“事件”转换为“可 await 的 Task”，实现异步等待消息。

2.1 定义等待方法

在服务端或调用模块中定义：

private TaskCompletionSource<int> _tcs;public Task<int> WaitForNextPackageAsync()
{_tcs = new TaskCompletionSource<int>();return _tcs.Task; // 返回可 await 的 Task
}

外部模块调用 await WaitForNextPackageAsync() 时，会挂起等待
直到 _tcs.SetResult(result) 被触发

2.2 接收客户端消息并触发 TCS

public event Func<StringPackageInfo, Task<int>> OnPackageReceived;private async ValueTask HandlePackageAsync(IAppSession session, StringPackageInfo package)
{int result = 0;// 调用外部模块处理消息if (OnPackageReceived != null){result = await OnPackageReceived.Invoke(package);}// 完成 TaskCompletionSource，将结果返回给等待方_tcs?.TrySetResult(result);// 同时可以给客户端发送响应await session.SendAsync(Encoding.UTF8.GetBytes(result.ToString() + "\r\n"));
}

这里实现了 消息接收与处理逻辑解耦：

SuperSocket 只负责接收消息
外部模块处理业务逻辑
调用模块异步等待处理结果

2.3 外部模块处理逻辑示例

mainWindow.OnPackageReceived += async (package) =>
{int result = 0;switch (package.Key.ToUpper()){case "ADD":result = package.Parameters.Select(int.Parse).Sum();break;case "SUB":result = package.Parameters.Select(int.Parse).Aggregate((x, y) => x - y);break;case "MULT":result = package.Parameters.Select(int.Parse).Aggregate((x, y) => x * y);break;}return result; // 返回给 TaskCompletionSource
};

2.4 调用模块等待结果

int result = await mainWindow.WaitForNextPackageAsync();
Console.WriteLine($"处理结果: {result}");

外部模块就像同步等待一样获得了处理结果
实际上整个流程是 异步、非阻塞 的

3️⃣ 支持多客户端或多条命令

如果有多个客户端或希望同时处理多条消息，可以使用 队列管理 TCS：

private ConcurrentQueue<TaskCompletionSource<int>> _queue = new();public Task<int> WaitNextAsync()
{var tcs = new TaskCompletionSource<int>();_queue.Enqueue(tcs);return tcs.Task;
}private async ValueTask HandlePackageAsync(IAppSession session, StringPackageInfo package)
{int result = CalculateResult(package);if (_queue.TryDequeue(out var tcs))tcs.TrySetResult(result);await session.SendAsync(Encoding.UTF8.GetBytes(result.ToString() + "\r\n"));
}