如何改进PowerShell v5中ForEach-Parallel函数的实现？

问题背景 #

我发现了一个适用于PowerShell v5的ForEach-Parallel函数，正尝试对其进行改进。由于工作依赖的代码库仅支持v5版本，无法升级到PowerShell v7.x。

遇到的问题 #

• 并行网络命令效率低下：运行网络相关命令时，并行执行的时间节省效果微乎其微，怀疑是否受限于Windows 10的TCP栈？
• 示例运行报错：第二个使用Test-NetConnection的示例，运行前会输出错误：

  One or more errors occurred.
  

• 参数设计不合理：当前实现仅通过假设-ScriptBlock参数处于位置0来隐藏该参数，这种方式不够严谨。

当前函数代码 #

function Invoke-ForEachParallel {
    [CmdletBinding()]
    [Alias('ForEach-Parallel')]
    [Diagnostics.CodeAnalysis.SuppressMessageAttribute(
        "PSAvoidUsingAliases",
        '',
        Justification = 'The used verb makes the most sense in this case.'
    )]
    param (
        [Parameter(Mandatory, ValueFromPipeline)]
        [Array]$Items, #! 代码中未直接使用该参数，仅支持管道输入

        [Parameter(Mandatory, Position = 0)]
        [Alias ('Process')] #* 为兼容foreach语法的不合理实现方式
        [scriptblock]$ScriptBlock,

        [Parameter()]
        [Object[]]$ArgumentList,

        [Parameter()]
        [int]$MaxRunspaces = 16,

        [Parameter()]
        [int]$TimeOut = (60 * 60 * 1000) #TODO: 是否需要支持秒作为输入单位？
    )

    if ($Input) {
        try {
            # 创建用于脚本块的可选参数列表
            $Arguments  = ''
            $Parameters = ''

            if ($ArgumentList) {
                for ($Index = 0; $Index -lt $ArgumentList.Length; $Index++) {
                    $Arguments  += ", `$$Index"
                    $Parameters += " `$$Index"
                }
            }

            # 创建要执行的脚本块
            # - 包裹传入的脚本块，以便将参数（ArgumentList）和当前项（$_）一起传递
            # - 始终加载当前模块

            $ScriptText =
@"
[CmdletBinding()]
param (`$_$Arguments)

function Wrapper {
$ScriptBlock
}

Wrapper $Parameters
"@

            $SessionState = [System.Management.Automation.Runspaces.InitialSessionState]::CreateDefault()

            # 传入当前主机，让每个运行空间的输出显示在当前主机中
            $Pool = [runspacefactory]::CreateRunspacePool(1, $MaxRunspaces, $sessionState, $Host)
            $Pool.Open()

            # 为每个项创建新的运行空间
            $Runspaces      = @()
            $AsyncResults   = @()
            $Exceptions     = @()

            foreach ($Item in $Input) {
                $Runspace = [powershell]::create()
                $Runspace.RunSpacePool = $Pool
                $Runspaces += $Runspace

                $Runspace.Streams.Error.add_DataAdded({
                    Param (
                        [Object]$Sender,
                        [System.Management.Automation.DataAddedEventArgs]$e #!$e 用途未明确标注
                    )

                    foreach ($Item in $Sender.ReadAll()) {
                        throw "$($Item.Exception.Message)"
                    }
                })

                # 添加生成的脚本块，传入当前项和可选参数
                [void]$Runspace.AddScript($ScriptText)
                [void]$Runspace.AddArgument($Item)

                if ($ArgumentList) {
                    [void]$Runspace.AddParameters($ArgumentList)
                }

                # 传递Verbose参数
                [void]$Runspace.AddParameter('Verbose', $VerbosePreference -eq 'Continue')

                # 同步启动运行空间
                $AsyncResult    = $Runspace.BeginInvoke()
                $AsyncResults   += $AsyncResult

            }#end foreach

            # 等待所有运行空间完成
            for ($Index = 0; $Index -lt $AsyncResults.Length; $Index++) {
                $null = [System.Threading.WaitHandle]::WaitAll($AsyncResults[$Index].AsyncWaitHandle, $TimeOut)
            }

            # 获取每个运行空间的结果
            $Errors = @()

            for ($Index = 0; $Index -lt $AsyncResults.Length; $Index++) {
                $AsyncResult    = $AsyncResults[$Index]
                $Runspace       = $Runspaces[$Index]

                # 如有需要，以下属性可提供运行空间完成状态的详细信息
                # $runspace.InvocationStateInfo.State
                # $runspace.InvocationStateInfo.Reason

                try {
                    Write-Output ($Runspace.EndInvoke($AsyncResult))
                }
                catch {
                    # 收集所有错误，统一返回
                    $Errors += $_
                }

            }#end for

        }#end try

        finally {
            if ($Pool) {
                $Pool.Close()
            }
        }

        # 处理错误
        if ($Errors) {
            if ($Errors.Length -eq 1) {
                throw $Errors[0]
            }
            else {
                $Exceptions = [exception[]]($Errors).Exception
                $AllMessages = $([string]::Join("`n", $Exceptions.Message))
                throw (New-Object AggregateException -ArgumentList "发生一个或多个错误:`n$AllMessages",$Exceptions)
            }
        }

    }#endif Input

<#
.SYNOPSIS
    使用PowerShell运行空间实现并行foreach。

.DESCRIPTION
    使用PowerShell运行空间实现并行foreach。

.NOTES
    CodeblackNL的MIT许可证

.LINK
    https://github.com/CodeblackNL/chocolatey-oneget-provider/tree/master

.PARAMETER Items
    要为其执行脚本块的项。

.PARAMETER ScriptBlock
    要为每个项执行的脚本块。

.PARAMETER ArgumentList
    要传递给脚本块的对象参数数组。

.PARAMETER MaxRunspaces
    尝试并行运行的最大运行空间数。
    实际并行执行的运行空间数由运行时决定，例如受可用核心数限制。
    默认值为16。

.PARAMETER TimeOut
    等待每个运行空间完成的时间，单位为毫秒。
    默认值为1小时。

.EXAMPLE
    1..10 | foreach {Invoke-WebRequest -Uri www.bing.com | select StatusCode}

    1..10 | foreach-parallel {Invoke-WebRequest -Uri www.bing.com | select StatusCode}

    并行10次获取网站的状态码。

.EXAMPLE
    1..10 | foreach {Test-NetConnection www.bing.com -Port 443 | select TcpTestSucceeded}

    1..10 | foreach-parallel {Test-NetConnection www.bing.com -Port 443 | select TcpTestSucceeded}

    并行10次测试TCP端口是否可打开。

.EXAMPLE
    (1..10) | foreach {'Sleeping'; sleep -Seconds 1}

    (1..10) | foreach-parallel {'Sleeping'; sleep -Seconds 1}

    并行10次休眠1秒。

#>

}

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