使用Start-ThreadJob与ForEach-Object -Parallel实现生产者消费者模式异常

异常现象原因说明 #

1. 调用producer()方法卡死 #

PowerShell 自定义类的实例方法在跨 Runspace 调用时会触发 PowerShell 运行时的跨线程调度锁（ForEach-Object -Parallel 的每个并行执行块、Start-ThreadJob 启动的任务都运行在独立的 Runspace 中），多个并行线程同时调用同一个实例的方法时会出现锁竞争死锁。
直接访问实例的 queue 字段调用 Add() 方法时，是直接操作 .NET 原生的 BlockingCollection<> 对象，不需要走 PowerShell 的方法调度逻辑，不会触发额外的锁，因此可以正常运行。

2. 取消Start-Sleep注释后卡死 #

这是 PowerShell 7 版本中 ForEach-Object -Parallel 的已知实现缺陷：该命令在部分场景下不会等所有输入项的并行处理逻辑全部完成就提前执行后续代码。你添加1秒睡眠时刚好触发了这个缺陷：并行块里的 Add 操作还没全部执行完成，主线程就已经执行了 $this.queue.CompleteAdding()，未完成的 Add 操作会因为队列已标记完成被永久阻塞，消费端也会因为收不到完整数据陷入等待，最终导致死锁。

修复方案 #

你可以根据使用场景选择以下任意一种修复方式：

• 优先直接操作公开的 .NET 集合对象，不要在 ForEach-Object -Parallel 并行块、ThreadJob 中调用 PowerShell 自定义类的实例方法
• 若需要封装操作逻辑，改用 C# 定义的类来实现生产者消费者逻辑，C# 编译的类方法跨 Runspace 调用不会触发 PowerShell 的额外锁
• 避免使用 ForEach-Object -Parallel 实现对时序要求严格的并行生产逻辑，改用 Start-ThreadJob 启动多个生产者任务，主动等待所有生产者任务完成后再调用 CompleteAdding
• 若必须保留睡眠逻辑，可加入计数同步机制（比如 CountdownEvent），确认所有生产者都完成数据入队后再进入睡眠

修复后的参考代码 #

using namespace System.Collections.Concurrent
using namespace System.Threading

class TestProducerConsumer
{
    [int] $result = 0
    [BlockingCollection[int]] $queue = [BlockingCollection[int]]::new()

    [void] consumer() {
        $sum = 0
        $it = $this.queue.GetConsumingEnumerable()
        foreach( $i in $it ) { $sum += $i }
        $this.result = $sum
    }
    
    [void] Run() {
        $job = Start-ThreadJob { ($using:this).consumer() }
        # 用CountdownEvent同步所有生产者完成状态
        $countdown = [CountdownEvent]::new(10)
        1..10 | ForEach-Object -Parallel { 
            $queue = ($using:this).queue
            $cd = $using:countdown
            $queue.Add($_)
            $cd.Signal()
        }
        # 等待所有生产者入队完成后再执行后续操作
        $countdown.Wait()
        # 现在加睡眠不会触发死锁
        Start-Sleep -Seconds 1
        $this.queue.CompleteAdding()
        $job | Receive-Job -Wait
        $countdown.Dispose()
    }
}

$t = [TestProducerConsumer]::new(); $t.Run(); $t

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