PowerShell中如何用Runspace池实现ArrayList动态增项并发递归处理

现有遍历ArrayList的串行for循环会在执行过程中向列表动态新增项，且需要迭代处理所有新增项，串行实现运行正常，共得到12个数组元素，符合预期。尝试使用RunspacePool改造为并发执行时，使用线程安全的同步ArrayList存储数据，无处理延迟时结果符合预期，但启用Start-Sleep 1模拟实际业务延迟后，初始for循环会在运行空间完成新增项操作前退出，仅初始的5个元素被提交到Runspace池，最终仅得到10个元素，结果不符合预期。

串行实现参考代码 #

$array = [System.Collections.ArrayList]@(1, 2, 3, 4, 5)
Write-Host "Number of items in array before loop: $($array.Count)"
for ($i = 0; $i -lt $array.Count; $i++) {
    Write-Host "Counter: $i`tArray: $array"
    if ($array[$i] -in @(1, 2, 3, 4, 5)) {
        $array.Add($array[$i] + 3) | Out-Null
    }
}
Write-Host "Array: $array"
Write-Host "Number of items in array after loop: $($array.Count)"

存在问题的并发实现代码 #

$pool = [RunspaceFactory]::CreateRunspacePool(1, 10)
$pool.Open()
$runspaces = @()

$scriptblock = {
    Param ($i, $array)
    # Start-Sleep 1 # <------ 启用后输出结果会出现明显偏差
    Write-Output "$i value: $array"
    if ($i -in @(1, 2, 3, 4, 5)) {
        $array.Add($i + 3) | Out-Null
    }
}

$array = [System.Collections.ArrayList]::Synchronized(([System.Collections.ArrayList]$(1, 2, 3, 4, 5)))
Write-Host "Number of items in array before loop: $($array.Count)"
for ($i = 0; $i -lt $array.Count; $i++) {
    $runspace = [PowerShell]::Create().AddScript($scriptblock).AddArgument($array[$i]).AddArgument($array)
    $runspace.RunspacePool = $pool
    $runspaces += [PSCustomObject]@{ Pipe = $runspace; Status = $runspace.BeginInvoke() }
}

while ($runspaces.Status -ne $null) {
    $completed = $runspaces | Where-Object { $_.Status.IsCompleted -eq $true }
    foreach ($runspace in $completed) {
        $runspace.Pipe.EndInvoke($runspace.Status)
        $runspace.Status = $null
    }
}
Write-Host "array: $array"
Write-Host "Number of items in array after loop: $($array.Count)"
$pool.Close()
$pool.Dispose()

原并发实现采用一次性固定遍历的逻辑提交任务，循环条件$i -lt $array.Count仅在每次迭代瞬间读取数组长度，主线程提交完初始5个元素对应的任务后就会退出提交循环，不会等待已提交任务执行产生新元素，后续Runspace新增的元素不会被提交处理，最终导致结果缺失。

将一次性遍历提交的逻辑改为持续检测待处理元素+跟踪运行中任务的消费模式，直到所有元素都被提交处理、所有运行中任务执行完成且无新元素产生时再退出流程，核心实现要点：

• 保留线程安全的Synchronized ArrayList存储数据，避免多线程下集合修改异常
• 新增线程安全的字典结构记录已提交处理的元素索引，避免同一元素被重复提交
• 常驻循环持续检测数组新索引，发现未提交的元素就创建Runspace任务提交到池
• 退出条件同时校验「无正在运行的任务」「所有现有元素都已提交处理」两个规则
• 增加短休眠避免CPU空转占用过高资源

修正后的可运行代码如下：

$pool = [RunspaceFactory]::CreateRunspacePool(1, 10)
$pool.Open()
$runspaces = @()
# 线程安全字典，记录已提交处理的元素索引，避免重复提交
$submittedIndex = [System.Collections.Concurrent.ConcurrentDictionary[int, byte]]::new()

$scriptblock = {
    Param ($i, $array)
    Start-Sleep 1 # 模拟业务处理延迟，开启后逻辑依然正常
    Write-Output "Processing value: $i`tCurrent array: $array"
    if ($i -in @(1, 2, 3, 4, 5)) {
        $array.Add($i + 3) | Out-Null
    }
}

# 初始化线程安全数组
$array = [System.Collections.ArrayList]::Synchronized(([System.Collections.ArrayList]$(1,2,3,4,5)))
Write-Host "Number of items in array before loop: $($array.Count)"

while ($true) {
    # 提交所有未处理的索引对应任务
    for ($i = 0; $i -lt $array.Count; $i++) {
        if (-not $submittedIndex.ContainsKey($i)) {
            # 原子操作标记索引为已提交，防止多线程下重复提交
            if ($submittedIndex.TryAdd($i, 0)) {
                $runspace = [PowerShell]::Create().AddScript($scriptblock).AddArgument($array[$i]).AddArgument($array)
                $runspace.RunspacePool = $pool
                $runspaces += [PSCustomObject]@{ Pipe = $runspace; Status = $runspace.BeginInvoke() }
            }
        }
    }

    # 回收已完成的Runspace资源
    $completed = $runspaces | Where-Object { $_.Status.IsCompleted -eq $true }
    foreach ($runspace in $completed) {
        $runspace.Pipe.EndInvoke($runspace.Status)
        $runspace.Pipe.Dispose()
        $runspace.Status = $null
    }
    # 移除已完成的任务跟踪记录
    $runspaces = $runspaces | Where-Object { $_.Status -ne $null }

    # 满足退出条件：无运行中任务 + 所有现有元素都已提交处理
    if ($runspaces.Count -eq 0 -and $submittedIndex.Count -eq $array.Count) {
        break
    }

    # 短暂休眠降低CPU占用
    Start-Sleep -Milliseconds 100
}

Write-Host "Final array: $array"
Write-Host "Number of items in array after loop: $($array.Count)"
$pool.Close()
$pool.Dispose()

该实现支持任意层级的动态新增元素递归处理，无论业务处理延迟多长，最终结果都会和串行版本一致，得到12个元素的正确输出。

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