如何通过队列优化高频非即时操作执行，提升性能与稳定性？

你的原方案核心思路没问题，但确实存在空循环浪费资源、单条处理效率低的问题，以下是针对性的优化建议：

1. 用信号量替代轮询，避免空循环 #

原代码每隔1秒轮询一次队列，不管有没有元素都跑一遍，完全是浪费CPU。可以用SemaphoreSlim（适配异步场景）实现"有任务再唤醒处理"的逻辑：

• 往队列添加元素时，调用Release()唤醒处理线程；
• 处理线程在队列空的时候，调用WaitAsync()进入等待状态，直到有新任务到来。

2. 批量处理，提升IO/网络效率 #

既然即时性要求不高，没必要逐条处理。可以攒一批任务（比如攒50条，或等待最多2秒）再一次性执行写入/上传操作，大幅减少IO和网络调用次数，提升整体性能。

3. 增加优雅停止机制 #

原代码是无限死循环，程序退出时没法正常收尾（比如队列里剩余的任务没处理完）。可以用CancellationToken实现优雅停止，收到停止信号时，处理完队列剩余任务再退出。

4. 增加异常容错处理 #

单个任务处理失败（比如网络波动、本地IO错误）时，不能让整个处理线程挂掉。要给处理逻辑加try-catch，甚至可以把失败的任务重新入队（注意加重试次数限制，避免死循环）。

优化后的示例代码 #

// 信号量，初始0表示无任务时等待
private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(0);
private readonly ConcurrentQueue<LogModel> _logQueue = new ConcurrentQueue<LogModel>();
private readonly CancellationTokenSource _cts = new CancellationTokenSource();

// 启动处理线程
public void StartLogProcessor()
{
    Task.Run(async () => await ProcessLogsAsync(), _cts.Token);
}

// 优雅停止
public void StopLogProcessor()
{
    _cts.Cancel();
    _semaphore.Release(); // 唤醒等待的线程，让它检测到取消信号
}

// 外部调用的添加日志方法
public void AddLog(LogModel log)
{
    _logQueue.Enqueue(log);
    _semaphore.Release(); // 唤醒处理线程
}

private async Task ProcessLogsAsync()
{
    var batch = new List<LogModel>(100); // 预设批量大小，减少内存分配
    while (!_cts.Token.IsCancellationRequested)
    {
        try
        {
            // 等待新任务，或最多等待2秒触发批量处理
            await _semaphore.WaitAsync(TimeSpan.FromSeconds(2), _cts.Token);

            // 尽可能取出队列元素攒成批量
            while (_logQueue.TryDequeue(out var log))
            {
                batch.Add(log);
                // 达到批量阈值立即处理
                if (batch.Count >= 100)
                {
                    await ProcessBatchAsync(batch);
                    batch.Clear();
                }
            }

            // 队列空但批量有元素，也处理掉
            if (batch.Count > 0)
            {
                await ProcessBatchAsync(batch);
                batch.Clear();
            }
        }
        catch (OperationCanceledException)
        {
            // 收到取消信号，退出循环
            break;
        }
        catch (Exception ex)
        {
            // 记录异常，避免线程挂掉
            Console.WriteLine($"日志处理失败: {ex.Message}");
            // 可选：将失败的任务重新入队（需加重试次数限制）
            foreach (var log in batch)
            {
                _logQueue.Enqueue(log);
            }
            batch.Clear();
            // 短暂等待避免异常风暴
            await Task.Delay(1000);
        }
    }

    // 退出前处理剩余任务
    if (batch.Count > 0)
    {
        try
        {
            await ProcessBatchAsync(batch);
        }
        catch (Exception ex)
        {
            Console.WriteLine("退出时处理剩余日志失败: " + ex.Message);
        }
    }
}

// 批量处理逻辑（替换为你的存储/上传代码）
private async Task ProcessBatchAsync(List<LogModel> batch)
{
    // 示例：批量写入本地日志
    var logLines = batch.Select(log => $"[{DateTime.Now:yyyy-MM-dd HH:mm:ss}] {log.Content}");
    await File.AppendAllLinesAsync("logs.txt", logLines);

    // 或批量上传WebAPI
    // await _httpClient.PostAsJsonAsync("/api/logs/batch", batch);
}

额外建议 #

• 如果是日志场景，直接用成熟框架（比如Serilog、NLog），它们内置批量写入、异步处理功能，不用自己造轮子；
• 控制批量大小和等待时间的平衡：批量太小起不到优化作用，太大可能导致延迟过高；
• 给队列加容量限制，避免任务堆积过多导致内存溢出。

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