Worker应用Release版本任务无异常中途停止问题求助

问题背景 #

开发部署在服务器上的基础Worker应用：从SQL Server获取任务，加入BlockingCollection队列，启动配置数量的Task消费队列任务。本地Debug版本运行正常，但Release版本中任务启动后速度更快，运行一段时间后会中途停止，无异常抛出；此时每个任务已生成20-80个文件，应用仍高占用CPU，但不再生成文件，等待半小时也无法恢复。

原核心代码 #

static void Main()
{
    int NumberOfConcurrentJobs = Convert.ToInt32(ConfigurationManager.AppSettings["NumberOfConcurrentJobs"]);
    int CollectionLimit = Convert.ToInt32(ConfigurationManager.AppSettings["MaxNumberOfQueueItems"]);

    /* Blocking Collection for the jobs to be consumed */
    BlockingCollection<Classes.Job> blockingCollection = new BlockingCollection<Classes.Job>(new ConcurrentQueue<Classes.Job>(), CollectionLimit);
    /* Blocking Collection to hold IDs for each Consumer Task so that they can be identified */
    BlockingCollection<int> ConsumerIDCollection = new BlockingCollection<int>(NumberOfConcurrentJobs);

    /* Start the Producer task */
    Task.Run(() =>
    {
        while (true)
        {
            /* 任务入队逻辑（确认正常，此处省略） */
            Thread.Sleep(2000); // 避免过于频繁填充队列
        }
    });

    /* Start the Consumer tasks */
    for (int i = 0; i < NumberOfConcurrentJobs; i++)
    {
        ConsumerIDCollection.Add(i + 1);
        /* Launch a task for each consumer */
        Task.Run(() =>
        {
            int ConsumerID = ConsumerIDCollection.Take();
            /* Loop forever, attempting to take a Job from the collection */
            while (true)
            {
                if (blockingCollection.TryTake(out Classes.Job job))
                {
                    try
                    {
                        Console.WriteLine(("(W) Consumer " + ConsumerID + ": Job " + job.JobID.ToString() + " taken...").PadRight(50) + "Processing.");
                        // 执行任务：混合CPU/IO操作，生成本地文件
                        job.RunWorker(); 
                        Console.WriteLine(("(W) Consumer " + ConsumerID + ": Job " + job.JobID.ToString() + " finished...").PadRight(50) + "Status " + job.Status + ".");
                    }
                    catch (Exception ex)
                    {
                        Common.WriteErrorLog(Common.LogType.Worker, "Consumer" + ConsumerID.ToString(), ex.Message);
                    }
                }

                Thread.Sleep(2000); // 等待后尝试获取下一个任务
            }
        });
    }
    Console.ReadKey();
}

注：job.RunWorker()为同步void方法，所有操作均同步执行；测试场景为4个并发Task各执行生成100个PDF文件的任务，每个任务在独立目录生成文件。

已验证测试场景 #

• 单任务生成100个PDF：可正常完成
• 并发多个小任务：均能正常消费并完成
• 在job.RunWorker()中手动抛出测试异常：可被外层try/catch捕获
• 尝试将消费Task存入数组并调用Task.WaitAll()替代Console.ReadKey()：问题未改变

排查方向建议 #

1. 未捕获的静默异常/线程终止 #

• 非托管代码异常：检查job.RunWorker()中调用的PDF生成组件是否存在原生代码异常，这类异常在Release模式下可能绕过托管代码的try/catch，直接终止线程。可添加全局TaskScheduler.UnobservedTaskException事件监听，或在每个Task.Run的委托外层再套一层try/catch，捕获所有未处理异常。
• 栈溢出异常：Release模式JIT优化可能导致栈溢出表现异常，不会抛出常规异常直接终止线程。可拆分PDF生成的大循环、减少深层调用，或临时增加栈空间测试。

2. 线程死锁/资源阻塞 #

• 共享资源未释放：检查PDF生成过程中是否存在全局文件句柄、数据库连接等共享资源未正确释放，导致所有线程卡在自旋等待状态（表现为CPU高但无输出）。
• 任务卡住定位：在job.RunWorker()前后添加包含线程ID、时间戳的详细日志，定位具体哪个任务、哪一步卡住。

3. JIT优化导致逻辑异常 #

• 临时关闭Release模式的代码优化（项目属性→生成→取消勾选“优化代码”），测试是否复现问题，确认是否为JIT优化导致逻辑偏差。
• 修正ConsumerID传递方式：原代码依赖ConsumerIDCollection获取ID，可直接将i+1作为参数传递给Task，避免Task调度顺序带来的潜在问题。

4. 资源耗尽问题 #

• 磁盘IO瓶颈：监控服务器磁盘空间、IO使用率，确认是否因磁盘满、临时文件未清理导致无法创建新文件。
• 线程池耗尽：若job.RunWorker()中大量使用Task.Run，可能耗尽线程池，导致后续任务无法调度。可监控线程数，或直接执行同步代码（无需异步包装）。

补充：ActionBlock实现的问题点及修正 #

用户提供的ActionBlock代码存在几个关键问题，可能导致任务停止：

1. 生产者循环中错误调用workerBlock.Complete()：会标记数据块停止接受新任务，后续SendAsync失败；应仅在应用停止时调用。
2. OnErrorLog为async void方法：异常无法被捕获，应改为async Task并等待执行。
3. job.RunWorker()用Task.Run包装同步代码：浪费线程池资源，直接执行同步逻辑即可。

修正后的ActionBlock示例代码 #

ActionBlock<Job> workerBlock = new ActionBlock<Job>(job =>
{
    Console.WriteLine($"{job.JobID} started...");
    try
    {
        job.RunWorker(); // 直接执行同步方法
    }
    catch (Exception ex)
    {
        Console.WriteLine(ex.Message);
        Common.WriteErrorLog(Common.LogType.Worker, job.JobID.ToString(), ex.Message);
    }
    finally
    {
        Console.WriteLine($"{job.JobID} done...");
    }
},
new ExecutionDataflowBlockOptions
{
    MaxDegreeOfParallelism = NumberOfConcurrentJobs,
    BoundedCapacity = QueueLimit
});

// 异步监控块完成情况
_ = MonitorBlockCompletion(workerBlock);

// 生产者循环
while (true)
{
    if (workerBlock.InputCount < QueueLimit)
    {
        List<int> JobIDs = ApiAction.GetJobsForWorker(QueueLimit);
        foreach (int JobID in JobIDs)
        {
            // 非阻塞发送，避免Wait()导致线程阻塞
            _ = workerBlock.SendAsync(new Job(JobID));
        }
    }
    Thread.Sleep(2000); // 避免频繁查询数据库
}

// 异步监控块完成情况的方法
public static async Task MonitorBlockCompletion(IDataflowBlock block)
{
    try
    {
        await block.Completion.ConfigureAwait(false);
    }
    catch (AggregateException ex)
    {
        foreach (var innerEx in ex.InnerExceptions)
        {
            Console.WriteLine($"{block.GetType().Name} failed: {innerEx.Message}");
            Common.WriteErrorLog(Common.LogType.Worker, "DataflowBlock", innerEx.Message);
        }
    }
    catch (Exception ex)
    {
        Console.WriteLine($"{block.GetType().Name} failed: {ex.Message}");
        Common.WriteErrorLog(Common.LogType.Worker, "DataflowBlock", ex.Message);
    }
}

// 修正后的Job类
class Job
{
    public void RunWorker() // 同步方法，无需异步包装
    {
        // 文件生成逻辑
    }
}

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