class Job {
    Init() {
        $this.ps.Streams.Information.add_DataAdded( {
            Param ( [Object]$sender, [System.Management.Automation.DataAddedEventArgs]$e )
    $i=0
            while(!($all = $sender.ReadAll())) {$i++;Write-Warning [Information] ReadAll阻塞：$i;Start-Sleep -Milliseconds 10}
            foreach($o in $all) { Write-Host $o }
        })

    }

}
复制代码

class Job {
    Init() {
        $this.ps.Streams.Information.add_DataAdded( {
            Param ( [Object]$sender, [System.Management.Automation.DataAddedEventArgs]$e )
            foreach($o in $sender.ReadAll()) { Write-Host $o }
        })
    }

}
复制代码

cls
# 处理任务
$Worker = {
    param($Filename)
    $i=0
    Write-Host "Processing $filename"
    $i=1
    #[Threading.Thread]::Sleep(100)
    Start-Sleep -Seconds 1 # Doing some work....
    $i=2
    Write-Host "Processed $filename"
    $i=3
}
# 最大并发
$MaxRunspaces = 10

try {
	# 线程池
    $RunspacePool = [runspacefactory]::CreateRunspacePool(1, $MaxRunspaces)
    $RunspacePool.Open()

    $Filenames = @("file1.txt", "file2.txt", "file3.txt", "file4.txt", "file5.txt", "file6.txt", "file7.txt", "file8.txt", "file9.txt", "file10.txt", "file11.txt")

    while($true) {
        # 开始时间
        $StartTime = Get-Date
        $Jobs = New-Object System.Collections.ArrayList

        foreach ($File in $Filenames) {
            Write-Host "Creating runspace for $File"
            $job=[Job]::new($RunspacePool)
            $null=$job.GetPowerShell().AddScript($Worker).AddArgument($File)
            $null=$Jobs.Add($job)
            $job.Run()
        }

        while ($Jobs.IsCompleted() -contains $false) {
            Write-Host (Get-date).Tostring() "Still running..."
            #[Threading.Thread]::Sleep(1000)
            Start-Sleep 1
            $Jobs.IsCompleted()
        }
        # 结束时间
        $endTime = Get-Date
        # 运行时间
        Write-Host "The script runs for Total time (s)：$(($endTime - $StartTime).TotalSeconds)"
        # 任务对象信息
        $Jobs.GetPowerShell().Streams.Information
        $Jobs.GetPowerShell().haderrors
        $c=Read-Host '输入e退出'
        if($c -eq 'e') {break}
    }
}
finally {
	# 始终关闭进程池
    $RunspacePool.Close()
}


class Job {
    hidden [powershell]$ps = [powershell]::Create()
    hidden [IAsyncResult]$result
    
    Job([System.Management.Automation.Runspaces.RunspacePool]$runspacepool) {
        $this.ps.RunspacePool = $runspacepool
        $this.Init()
    }

    Init() {
        $this.ps.Streams.Information.add_DataAdded( {
            Param ( [Object]$sender, [System.Management.Automation.DataAddedEventArgs]$e )
            while(!($all = $sender.ReadAll())) {Start-Sleep -Milliseconds 10}
            foreach($o in $all) { Write-Host $o }
        })
    }

    [powershell] GetPowerShell() {
        return $this.ps
    }
    
    [IAsyncResult] GetResult() {
        return $this.result
    }

    [bool]IsCompleted() {
        return $this.result.IsCompleted
    }

    [IAsyncResult] Run() {
        return $this.result = $this.ps.BeginInvoke()
    }
}
复制代码

workflow work1 {
    $Filenames = @("file1.txt", "file2.txt", "file3.txt", "file4.txt", "file5.txt", "file6.txt", "file7.txt", "file8.txt", "file9.txt", "file10.txt", "file11.txt")
  # workflow有很多限制
  # foreach -parallel允许同时最多开启5个线程
  foreach -parallel ($Filename in $Filenames) {
    $i=0
     "Processing $filename"
    $i=1
    #[Threading.Thread]::Sleep(100)
    Start-Sleep -Seconds 1 # Doing some work....
    $i=2
     "Processed $filename"
    $i=3
  }
  
}
[System.Diagnostics.Stopwatch]$watch = [System.Diagnostics.Stopwatch]::StartNew()
work1
$watch.Stop()
$watch.Elapsed # 耗时2.21秒
复制代码

#Runspace最大数
$MaxRunspaces = 10

#Runspace池
$RunspacePool = [runspacefactory]::CreateRunspacePool(1, $MaxRunspaces)
$RunspacePool.Open()

#任务
$Worker = {
param($Filename)
if(test-path $Filename){
$all = gc $Filename
foreach($o in $all) { [console]::writeline($o)}
$rt=0
}else{
$rt=-1
}
$rt
}

$Filenames = @("file1.txt", "file2.txt", "file3.txt", "file4.txt", "file5.txt", "file6.txt", "file7.txt", "file8.txt", "file9.txt", "file10.txt", "file11.txt")

$jobs = @()

#添加并执行任务
foreach ($File in $Filenames) {
$ps = [Powershell]::Create()
$job = $ps.AddScript($Worker).AddArgument($File)
$job.RunspacePool = $RunspacePool
$jobs += New-Object PSObject -Property @{ 
Job = $job
PowerShell = $ps
Result = $job.BeginInvoke() 
}
}

#判断任务是否完成
do{
Start-Sleep -Milliseconds 500
$c = ($jobs | Where {$_.Result.IsCompleted -ne $true}).Count
Write-Host ("未完成任务数" + $c) -ForegroundColor Green
}while($c -gt 0)

#获取返回结果并释放资源
$i=1;foreach($j in $jobs){
$r = $j.Job.EndInvoke($j.Result)
write-host "任务 $i 返回值 : $r" -ForegroundColor Yellow
$j.PowerShell.Dispose()
$i+=1
}

cmd /c pause
复制代码