在Parallel.ForEach中检测状态，实现OnStop()时等待任务完成再停止Windows服务

嘿，我来帮你搞定这个需求！要让Windows服务在调用OnStop()时等待所有Parallel.ForEach的线程完成后再停止，核心是用**取消令牌（CancellationToken）**来控制并行操作，同时配合同步机制和任务跟踪来确保优雅关闭。下面是具体的实现方案，结合你的示例代码改造：

第一步：添加必要的成员变量 #

首先在你的服务类里新增几个核心变量，用来控制取消信号、跟踪并行任务和同步操作：

private Timer _timer;
private CancellationTokenSource _cts;
private readonly object _lockObj = new object(); // 防止Timer重复触发并行操作
private Task _currentParallelTask; // 跟踪当前正在执行的并行任务

第二步：初始化服务与计时器 #

在OnStart方法里初始化取消令牌和计时器，设置触发并行操作的间隔：

protected override void OnStart(string[] args)
{
    _cts = new CancellationTokenSource();
    // 示例：立即执行一次，之后每5分钟触发一次（根据你的需求调整间隔）
    _timer = new Timer(ExecuteParallelWork, null, TimeSpan.Zero, TimeSpan.FromMinutes(5));
}

第三步：改造并行操作的执行逻辑 #

把原来的Parallel.ForEach包装在Task里，并且传入取消令牌，确保它能响应停止信号：

private void ExecuteParallelWork(object state)
{
    lock (_lockObj)
    {
        // 如果已经收到取消信号，直接返回，不启动新的并行任务
        if (_cts.Token.IsCancellationRequested)
            return;

        // 把Parallel.ForEach包装成Task，方便后续等待完成
        _currentParallelTask = Task.Run(() =>
        {
            // 替换成你实际要处理的集合
            var itemsToProcess = GetYourItems();

            var parallelOptions = new ParallelOptions
            {
                CancellationToken = _cts.Token,
                // 可选：设置最大并行度，避免占用过多系统资源
                MaxDegreeOfParallelism = Environment.ProcessorCount
            };

            try
            {
                Parallel.ForEach(itemsToProcess, parallelOptions, item =>
                {
                    // 每次迭代前检查取消信号，尽早响应停止请求
                    parallelOptions.CancellationToken.ThrowIfCancellationRequested();

                    // 执行你的具体业务操作
                    ProcessSingleItem(item);
                });
            }
            catch (OperationCanceledException)
            {
                // 取消信号触发的异常，属于正常停止，可记录日志或忽略
                EventLog.WriteEntry("ScheduledService", "Parallel operation cancelled gracefully.", EventLogEntryType.Information);
            }
        }, _cts.Token);
    }
}

// 示例：你的业务处理方法，建议也加入取消检查
private void ProcessSingleItem(object item)
{
    // 耗时操作前检查取消信号
    _cts.Token.ThrowIfCancellationRequested();
    
    // 执行第一步操作
    DoBusinessLogicStep1(item);

    _cts.Token.ThrowIfCancellationRequested();
    
    // 执行第二步操作
    DoBusinessLogicStep2(item);
}

// 示例：获取待处理集合的方法
private IEnumerable<object> GetYourItems()
{
    // 替换成你实际获取数据的逻辑
    return new List<object> { /* 你的数据 */ };
}

第四步：实现优雅的OnStop逻辑 #

在OnStop方法里，依次完成停止计时器、触发取消、等待任务完成、清理资源这几个步骤：

protected override void OnStop()
{
    // 关键：请求额外的停止时间（默认服务停止超时约15秒，根据你的任务耗时调整）
    // 示例：请求5分钟的额外时间，单位是毫秒
    RequestAdditionalTime(300000);

    // 1. 停止计时器，避免触发新的并行任务
    _timer?.Change(Timeout.Infinite, Timeout.Infinite);

    // 2. 触发取消信号，通知并行操作停止
    _cts?.Cancel();

    // 3. 等待当前正在运行的并行任务完成
    try
    {
        lock (_lockObj)
        {
            _currentParallelTask?.Wait(_cts.Token);
        }
    }
    catch (AggregateException ex)
    {
        // 过滤掉取消异常，只处理真正的业务错误
        var nonCancelExceptions = ex.Flatten().InnerExceptions
                                   .Where(e => !(e is OperationCanceledException));
        
        if (nonCancelExceptions.Any())
        {
            foreach (var exception in nonCancelExceptions)
            {
                EventLog.WriteEntry("ScheduledService", $"Error during service stop: {exception.Message}", EventLogEntryType.Error);
            }
        }
    }
    catch (OperationCanceledException)
    {
        // 取消异常，属于正常流程，无需额外处理
    }

    // 4. 清理资源
    _currentParallelTask?.Dispose();
    _cts?.Dispose();
    _timer?.Dispose();

    EventLog.WriteEntry("ScheduledService", "Service stopped gracefully.", EventLogEntryType.Information);
}

关键注意点 #

• 取消令牌的作用：Parallel.ForEach会在每次迭代开始前检查取消信号，不会强制终止正在执行的迭代（除非你手动在业务方法里检查令牌并抛出异常），确保正在处理的任务能完成当前操作后再退出。
• 锁的必要性：防止Timer在并行任务未完成时再次触发，导致多个Parallel.ForEach并发运行。
• 超时处理：一定要调用RequestAdditionalTime，如果你的并行任务需要较长时间完成，避免服务被系统强制终止。
• 异常处理：区分取消异常和业务异常，只记录真正的错误。

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