C#多线程异步请求与响应乱序场景的同步方案咨询

问题背景 #

多个独立线程通过GetHistoryData(securityCode)发起请求，响应由_handleData线程异步返回。要求每个响应必须与对应请求关联，但响应返回顺序可能和请求顺序不一致。原代码使用AutoResetEvent实现同步时，出现等待线程接收到其他线程响应的错误，单线程正常但多线程场景失效。

问题根源 #

原代码的核心问题是所有请求共享同一个全局AutoResetEvent和_candles变量：

• 当任意一个响应触发_candlesLoaded.Set()时，第一个等待的线程会被唤醒，但此时_candles可能已经被其他响应覆盖，导致线程拿到错误的结果。
• 全局变量的无保护读写在多线程场景下会引发竞态条件，进一步加剧数据混乱。

解决方案：基于请求标识的线程隔离同步 #

核心思路是为每个请求分配唯一标识，用线程安全字典存储每个请求对应的同步对象和结果，确保响应只唤醒对应请求的等待线程。

1. 定义请求上下文类 #

用于存储每个请求的同步事件和结果：

private class RequestContext
{
    public AutoResetEvent WaitEvent { get; set; }
    public List<candle> Result { get; set; }
}

2. 初始化线程安全字典 #

存储请求ID与对应上下文的映射：

private readonly ConcurrentDictionary<string, RequestContext> _requestContexts = new ConcurrentDictionary<string, RequestContext>();

3. 修改GetHistoryData方法 #

为每个请求生成唯一ID，绑定专属同步事件，等待自身响应：

public List<candle> GetHistoryData(string seccode, boardsCode board = boardsCode.FUT, SecurityPeriods periodId = SecurityPeriods.M1, int candlesCount = 1)
{
    // 生成唯一请求ID
    string requestId = Guid.NewGuid().ToString();
    var context = new RequestContext { WaitEvent = new AutoResetEvent(false) };
    _requestContexts.TryAdd(requestId, context);

    try
    {
        command com = command.CreateGetHistoryDataCommand(board, seccode, periodId, candlesCount);
        // 给命令附加请求ID，确保响应能关联到对应请求
        com.RequestId = requestId;
        result sendResult = ConnectorSendCommand(com, typeof(command));
        
        if (sendResult.success)
        {
            // 仅等待当前请求的专属事件
            context.WaitEvent.WaitOne();
        }
        else
        {
            throw new Exception(sendResult.message);
        }

        return context.Result;
    }
    finally
    {
        // 清理资源，避免内存泄漏
        _requestContexts.TryRemove(requestId, out _);
        context.WaitEvent.Dispose();
    }
}

4. 修改_handleData方法 #

解析响应中的请求ID，找到对应上下文并唤醒等待线程：

protected override void _handleData(String result)
{
    string nodeName = _getNodeName(result);
    switch (nodeName)
    {
        case "candles":
            var candles = (candles)XMLHelper.Deserialize(result, typeof(candles));
            // 从响应中解析请求ID（需根据实际XML结构实现_getRequestIdFromResult方法）
            string requestId = _getRequestIdFromResult(result); 
            
            if (_requestContexts.TryGetValue(requestId, out var context))
            {
                context.Result = candles.candlesValue;
                // 仅触发对应请求的等待事件
                context.WaitEvent.Set();
            }
            break;
    }
}

适用设计模式 #

• Command模式：将每个请求封装为command对象，附加请求ID后可完整跟踪命令的发起与响应，符合请求-响应的解耦需求。
• Future模式：每个请求的RequestContext相当于Future实现，请求线程等待Future结果，异步响应线程填充结果并唤醒等待者，实现异步结果的同步获取。
• 生产者-消费者模式：请求线程作为生产者提交命令，_handleData线程作为消费者处理响应，线程安全字典作为缓冲区关联两者，确保响应准确分发。

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