C#定时队列HTTP调用类如何实现可等待写入完成的WriteAsync异步方法

实现方案 #

你原来的实现存在几个核心问题：

• ManualResetEvent.WaitOne()是同步阻塞逻辑，即便写法上返回了Task，实际上还是会占用线程池线程等待，并没有实现真正的异步
• 未处理ManualResetEvent的资源释放，容易造成资源泄露
• 没有异常传播能力，HTTP调用失败时调用方无法感知，也没有重试逻辑
• 未做队列持久化，进程崩溃或重启时未发送的请求会直接丢失

下面是优化后的实现方案，基于TaskCompletionSource实现纯异步等待，同时支持队列持久化保证请求不丢失：

完整代码实现 #

public class AsyncValueTimerIntervalWriter
{
    private class ValueRequest
    {
        public string FullName { get; set; }
        public object Value { get; set; }
        // 用于异步通知完成状态
        public TaskCompletionSource<bool> Tcs { get; set; }
    }

    private readonly IValuesClient _valuesClient;
    private List<ValueRequest> _valueRequests = new List<ValueRequest>();
    private readonly object _valuesLock = new object();
    private Timer _timer;
    private Task _runningWriteTask;
    private readonly CancellationTokenSource _stopCts = new CancellationTokenSource();

    public AsyncValueTimerIntervalWriter(IValuesClient valuesClient)
    {
        _valuesClient = valuesClient;
    }

    public void Start()
    {
        lock (_valuesLock)
        {
            // 若需要跨进程重启不丢请求，此处从本地持久化存储加载历史未发送的请求
            // 可选用LiteDB、LevelDB等轻量嵌入式数据库，或本地JSON文件实现持久化
            // var persistedRequests = LoadRequestsFromLocalStorage();
            // _valueRequests.AddRange(persistedRequests);
        }

        _timer = new Timer(async _ =>
        {
            // 上一批次未写完则跳过本次执行，避免并发写入
            if (_runningWriteTask != null && !_runningWriteTask.IsCompleted)
                return;

            _runningWriteTask = WriteValuesToServer();
            await _runningWriteTask;
        }, null, TimeSpan.Zero, TimeSpan.FromSeconds(1));
    }

    public async Task StopAsync()
    {
        _timer?.Dispose();
        _stopCts.Cancel();

        // 等待当前正在执行的写入任务完成，避免丢失正在处理的请求
        if (_runningWriteTask != null)
            await _runningWriteTask;

        lock (_valuesLock)
        {
            // 停止时将未发送的请求持久化到本地，保证重启后可恢复
            // SaveRequestsToLocalStorage(_valueRequests);
        }
    }

    public void AddWrite(string fullName, object value)
    {
        _ = WriteAsync(fullName, value);
    }

    public Task WriteAsync(string fullName, object value)
    {
        var tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
        lock (_valuesLock)
        {
            _valueRequests.Add(new ValueRequest
            {
                FullName = fullName,
                Value = value,
                Tcs = tcs
            });
        }
        // 若开启持久化，此处新增的请求需要同时写入本地存储
        return tcs.Task;
    }

    private async Task WriteValuesToServer()
    {
        if (_stopCts.IsCancellationRequested)
            return;

        ValueRequest[] currentBatch;
        lock (_valuesLock)
        {
            currentBatch = _valueRequests.ToArray();
            _valueRequests.Clear();
        }

        if (currentBatch.Length == 0)
            return;

        try
        {
            await _valuesClient.SetValuesAsync(currentBatch);
            // 写入成功，通知所有等待的异步任务完成
            foreach (var req in currentBatch)
                req.Tcs?.TrySetResult(true);
            
            // 若开启持久化，此处删除本地存储中这批已成功发送的请求
        }
        catch (Exception ex)
        {
            // 写入失败，将请求塞回队列头部下次重试
            lock (_valuesLock)
            {
                _valueRequests.InsertRange(0, currentBatch);
            }

            // 若不需要自动重试，而是直接将错误抛给调用方，可以替换上面的逻辑为：
            // foreach (var req in currentBatch)
            //     req.Tcs?.TrySetException(ex);
        }
    }
}

核心特性说明 #

• 纯异步等待：基于TaskCompletionSource实现无阻塞等待，调用await WriteAsync时不会占用额外线程
• 异常处理：可选自动重试或者直接将异常传播给调用方，避免调用方无限等待
• 不丢请求保障：如果需要跨进程重启不丢请求，只要放开代码中持久化相关的注释实现即可，所有未发送的请求都会在停止时落地到本地存储，启动时重新加载到队列
• 并发安全：内置防并发写入逻辑，避免同一时间发起多个HTTP请求导致服务端压力过大

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