C#中接收特定外部消息后，如何在指定位置触发函数续行？

我们有一段C#代码：

public async Task TestMethod(){
  // 执行前置逻辑
  string a = await WaitingObj.Read();
  // 执行后续逻辑
}

我们维护着一个持续接收消息的基础队列，当队列收到属于TestMethod所需的消息时，需要把该消息传递给WaitingObj.Read()，让它返回消息并赋值给string a。同时要求不能调用Read()两次（一次在TestMethod中等待，一次在队列侧判断匹配）。

如何通过Async/Await或合理的程序设计解决这个问题？

核心思路是用TaskCompletionSource<T>实现异步等待的信号机制：让WaitingObj持有一个可被外部触发完成的任务，队列收到匹配消息时直接完成该任务，完全避免重复调用Read()。

1. 实现WaitingObj类 #

public class WaitingObj
{
    // 存储当前等待的异步任务，泛型对应返回值类型string
    private TaskCompletionSource<string> _taskSource;

    public Task<string> Read()
    {
        // 每次调用Read时创建新的任务源，确保单次等待对应单次消息
        _taskSource = new TaskCompletionSource<string>();
        return _taskSource.Task;
    }

    // 供消息队列调用，传入匹配消息完成等待任务
    public void DeliverMessage(string message)
    {
        _taskSource?.TrySetResult(message);
    }
}

2. 消息队列的处理逻辑 #

假设你的队列有一个持续监听的处理方法，在其中判断消息是否匹配需求，匹配时直接触发WaitingObj的任务完成：

public class MessageQueue
{
    private readonly Queue<string> _queue = new Queue<string>();
    private readonly WaitingObj _waitingObj;

    public MessageQueue(WaitingObj waitingObj)
    {
        _waitingObj = waitingObj;
    }

    public async Task StartProcessing()
    {
        while (true)
        {
            if (_queue.Count > 0)
            {
                var message = _queue.Dequeue();
                // 判断消息是否属于TestMethod的需求（替换成你的实际匹配逻辑）
                if (IsTargetMessage(message))
                {
                    // 直接传递消息，触发WaitingObj的等待任务完成
                    _waitingObj.DeliverMessage(message);
                }
                else
                {
                    // 处理其他非目标消息
                    ProcessOtherMessages(message);
                }
            }
            await Task.Delay(100); // 避免空转，实际可替换为队列的阻塞取消息逻辑
        }
    }

    private bool IsTargetMessage(string message)
    {
        // 示例匹配规则：消息以"TestMethod_"开头
        return message.StartsWith("TestMethod_");
    }

    private void ProcessOtherMessages(string message)
    {
        // 其他消息的处理逻辑
    }
}

3. 完整调用示例 #

public async Task RunDemo()
{
    var waitingObj = new WaitingObj();
    var queue = new MessageQueue(waitingObj);

    // 启动消息队列的后台处理
    _ = queue.StartProcessing();

    // 执行TestMethod
    await TestMethod(waitingObj);
}

public async Task TestMethod(WaitingObj waitingObj)
{
    Console.WriteLine("正在等待目标消息...");
    string a = await waitingObj.Read();
    Console.WriteLine($"收到目标消息：{a}");
    // 执行后续逻辑
}

逻辑说明 #

• WaitingObj.Read()仅在TestMethod中调用一次，它返回一个未完成的Task<string>，当前线程会异步等待该任务完成。
• 消息队列收到匹配消息时，直接调用waitingObj.DeliverMessage(message)，将消息传递给TaskCompletionSource并触发任务完成，此时await waitingObj.Read()会立即返回该消息，TestMethod继续执行后续代码。
• 每次调用Read()都会创建新的TaskCompletionSource，确保单次等待对应单次消息，不会出现重复触发的问题。

内容的提问来源于stack exchange，提问作者DamXosp4j