PowerShell，使用 Start-Job 和 Start-Process 测试异步任务的性能/效率

Question

我很想用 Start-ThreadJob 测试 PowerShell 中异步任务的性能/有用性| , Start-Job和 Start-Process .我有一个包含大约 100 个 zip 文件的文件夹，因此想出了以下测试:

New-Item "000" -ItemType Directory -Force   # Move the old zip files in here
foreach ($i in $zipfiles) {
    $name = $i -split ".zip"
    Start-Job -scriptblock {
        7z.exe x -o"$name" .\$name
        Move-Item $i 000\ -Force
        7z.exe a $i .\$name\*.*
    }
}

问题是它会启动所有 100 个 zip 的作业，这可能太多了，所以我想设置一个值 $numjobs ，比如 5，我可以更改，这样只有 $numjobs将同时启动，然后脚本将检查在下一个 5 block 开始之前结束的所有 5 个作业。然后我想根据 $numjobs 的值观察 CPU 和内存

我如何告诉一个循环只运行 5 次，然后等待作业完成再继续？

我看到等待作业完成很容易

$jobs = $commands | Foreach-Object { Start-ThreadJob $_ }
$jobs | Receive-Job -Wait -AutoRemoveJobchange

但是我怎么等Start-Process任务结束？

虽然我想用 Parallel-ForEach ，我工作的企业将在接下来的 3-4 年内与 PowerShell 5.1 紧密相关，我预计没有机会安装 PowerShell 7.x(尽管我很想在我的家庭系统上使用 Parallel-ForEach 进行测试比较所有方法)。

Answer 1

ForEach-Object -Parallel和 Start-ThreadJob具有限制可同时运行的线程数的内置功能，这同样适用于 Runspace和他们的 RunspacePool这是两个 cmdlet 在幕后使用的内容。

Start-Job不提供此类功能，因为每个作业都在单独的进程中运行，这与之前提到的在同一进程中的不同线程中运行的 cmdlet 不同。我个人也不认为它是并行的替代方案，它非常慢，而且在大多数情况下线性循环会比它快。 Serialization and deserialization在某些情况下也可能是个问题。

如何限制运行线程数？

这两个 cmdlet 都为此提供了 -ThrottleLimit 参数。

• https://learn.microsoft.com/en-us/powershell/module/threadjob/start-threadjob?view=powershell-7.2#-throttlelimit
• https://learn.microsoft.com/en-us/powershell/module/microsoft.powershell.core/foreach-object?view=powershell-7.2#-throttlelimit

代码看起来如何？

$dir = (New-Item "000" -ItemType Directory -Force).FullName

# ForEach-Object -Parallel
$zipfiles | ForEach-Object -Parallel {
    $name = [IO.Path]::GetFileNameWithoutExtension($_)
    7z.exe x -o $name .\$name
    Move-Item $_ $using:dir -Force
    7z.exe a $_ .\$name\*.*
} -ThrottleLimit 5

# Start-ThreadJob
$jobs = foreach ($i in $zipfiles) {
    Start-ThreadJob {
        $name = [IO.Path]::GetFileNameWithoutExtension($using:i)
        7z.exe x -o $name .\$name
        Move-Item $using:i $using:dir -Force
        7z.exe a $using:i .\$name\*.*
    } -ThrottleLimit 5
}
$jobs | Receive-Job -Wait -AutoRemoveJob

如何在只有 PowerShell 5.1 可用且无法安装新模块的情况下实现同样的目标？

RunspacePool提供相同的功能，或者是 .SetMaxRunspaces(Int32) Method或者通过定位 RunspaceFactory.CreateRunspacePool overloads 之一提供一个 maxRunspaces 限制作为参数。

代码看起来如何？

$dir   = (New-Item "000" -ItemType Directory -Force).FullName
$limit = 5
$iss   = [initialsessionstate]::CreateDefault2()
$pool  = [runspacefactory]::CreateRunspacePool(1, $limit, $iss, $Host)
$pool.ThreadOptions = [Management.Automation.Runspaces.PSThreadOptions]::ReuseThread
$pool.Open()

$tasks  = foreach ($i in $zipfiles) {
    $ps = [powershell]::Create().AddScript({
        param($path, $dir)

        $name = [IO.Path]::GetFileNameWithoutExtension($path)
        7z.exe x -o $name .\$name
        Move-Item $path $dir -Force
        7z.exe a $path .\$name\*.*
    }).AddParameters(@{ path = $i; dir = $dir })
    $ps.RunspacePool = $pool

    @{ Instance = $ps; AsyncResult = $ps.BeginInvoke() }
}

foreach($task in $tasks) {
    $task['Instance'].EndInvoke($task['AsyncResult'])
    $task['Instance'].Dispose()
}
$pool.Dispose()

请注意，对于所有示例，尚不清楚 7zip 代码是否正确，此答案试图演示如何在 PowerShell 中完成异步，而不是如何压缩文件/文件夹。

---

下面是一个辅助函数，可以简化并行调用的过程，尝试模拟 ForEach-Object -Parallel 并与 PowerShell 5.1 兼容，但 不应将其视为稳健的解决方案:

using namespace System.Management.Automation
using namespace System.Management.Automation.Runspaces
using namespace System.Collections.Generic

function Invoke-Parallel {
    [CmdletBinding()]
    param(
        [Parameter(Mandatory, ValueFromPipeline, DontShow)]
        [object] $InputObject,

        [Parameter(Mandatory, Position = 0)]
        [scriptblock] $ScriptBlock,

        [Parameter()]
        [int] $ThrottleLimit = 5,

        [Parameter()]
        [hashtable] $ArgumentList
    )

    begin {
        $iss = [initialsessionstate]::CreateDefault2()
        if($PSBoundParameters.ContainsKey('ArgumentList')) {
            foreach($argument in $ArgumentList.GetEnumerator()) {
                $iss.Variables.Add([SessionStateVariableEntry]::new($argument.Key, $argument.Value, ''))
            }
        }
        $pool  = [runspacefactory]::CreateRunspacePool(1, $ThrottleLimit, $iss, $Host)
        $tasks = [List[hashtable]]::new()
        $pool.ThreadOptions = [PSThreadOptions]::ReuseThread
        $pool.Open()
    }
    process {
        try {
            $ps = [powershell]::Create().AddScript({
                $args[0].InvokeWithContext($null, [psvariable]::new("_", $args[1]))
            }).AddArgument($ScriptBlock.Ast.GetScriptBlock()).AddArgument($InputObject)

            $ps.RunspacePool = $pool
            $invocationInput = [PSDataCollection[object]]::new(1)
            $invocationInput.Add($InputObject)

            $tasks.Add(@{
                Instance    = $ps
                AsyncResult = $ps.BeginInvoke($invocationInput)
            })
        }
        catch {
            $PSCmdlet.WriteError($_)
        }
    }
    end {
        try {
            foreach($task in $tasks) {
                $task['Instance'].EndInvoke($task['AsyncResult'])
                if($task['Instance'].HadErrors) {
                    $task['Instance'].Streams.Error
                }
                $task['Instance'].Dispose()
            }
        }
        catch {
            $PSCmdlet.WriteError($_)
        }
        finally {
            if($pool) { $pool.Dispose() }
        }
    }
}

它是如何工作的一个例子:

# Hashtable Key becomes the Variable Name inside the Runspace!
$outsideVariables = @{ Message = 'Hello from {0}' }
0..10 | Invoke-Parallel {
    "[Item $_] - " + $message -f [runspace]::DefaultRunspace.InstanceId
    Start-Sleep 5
} -ArgumentList $outsideVariables -ThrottleLimit 3