如何将ForEach-Object -Parallel输出存入字典/ArrayList？分支重复问题排查

重复条目原因 #

最可能的错误是在ForEach-Object -Parallel块中引用外部线程安全字典时未使用$using:前缀，导致每个并行线程创建了独立的字典实例，最终合并多个字典的结果时出现重复条目。此外，还可能存在以下问题：

• 未使用分支的唯一标识（如仓库ID+分支名）作为字典键，不同仓库的同名分支被误判为重复
• 并行任务未拆分数据源，多个线程重复处理同一批分支集合

修正方案 #

1. 正确使用ConcurrentDictionary与$using:前缀 #

确保并行线程共享同一个线程安全字典，并使用唯一键避免重复：

# 初始化线程安全字典
$staleBranches = [System.Collections.Concurrent.ConcurrentDictionary[string, PSObject]]::new()

$orgUrl = "https://dev.azure.com/your-org"
$projectName = "your-project"
$repositories = az repos list --organization $orgUrl --project $projectName --output json | ConvertFrom-Json

# 并行处理每个仓库
$repositories | ForEach-Object -Parallel {
    # 用$using:引用外部字典和变量
    $dict = $using:staleBranches
    $org = $using:orgUrl
    $project = $using:projectName
    $cutoffDate = (Get-Date).AddMonths(-3)

    # 获取当前仓库的分支
    $branches = az repos list --repository $_.id --organization $org --project $project --output json | ConvertFrom-Json

    foreach ($branch in $branches) {
        if ([DateTime]$branch.lastCommit.date -lt $cutoffDate) {
            # 仓库ID+分支名作为唯一键，避免跨仓库同名分支冲突
            $uniqueKey = "$($_.id)-$($branch.name)"
            # 尝试添加，已存在则自动跳过
            $dict.TryAdd($uniqueKey, $branch)
        }
    }
} -ThrottleLimit 8 # 根据机器性能调整并发数

# 输出去重后的陈旧分支
$staleBranches.Values | Select-Object name, repositoryId, @{Name="LastCommitDate"; Expression={[DateTime]$_.lastCommit.date}}

2. 其他替代实现方法 #

方法一：优化串行脚本（减少请求开销）

如果并行处理的复杂度太高，可通过批量调用Azure DevOps REST API减少请求次数，提升串行脚本速度：

$orgUrl = "https://dev.azure.com/your-org"
$projectName = "your-project"
$pat = "your-personal-access-token"
$headers = @{Authorization = "Basic " + [Convert]::ToBase64String([Text.Encoding]::ASCII.GetBytes(":$pat"))}

# 获取所有仓库
$repos = Invoke-RestMethod -Uri "$orgUrl/$projectName/_apis/git/repositories?api-version=7.1-preview.1" -Headers $headers | Select-Object -ExpandProperty value

$staleBranches = @()
foreach ($repo in $repos) {
    # 获取当前仓库的所有分支（过滤heads分支）
    $branches = Invoke-RestMethod -Uri "$orgUrl/$projectName/_apis/git/repositories/$($repo.id)/refs?filter=heads&api-version=7.1-preview.1" -Headers $headers | Select-Object -ExpandProperty value
    $cutoffDate = (Get-Date).AddMonths(-3)
    # 过滤陈旧分支
    $stale = $branches | Where-Object { [DateTime]$_.lastCommit.date -lt $cutoffDate }
    $staleBranches += $stale | Select-Object name, @{Name="RepoId"; Expression={$repo.id}}, @{Name="LastCommitDate"; Expression={[DateTime]$_.lastCommit.date}}
}

方法二：使用Start-ThreadJob实现并行

通过线程作业更灵活地控制并行任务：

Import-Module Az.DevOps

$orgUrl = "https://dev.azure.com/your-org"
$projectName = "your-project"
Connect-AzDevOps -Organization $orgUrl -Project $projectName

$repos = Get-AzDevOpsRepository
$jobs = @()
$cutoffDate = (Get-Date).AddMonths(-3)

foreach ($repo in $repos) {
    $jobs += Start-ThreadJob -ScriptBlock {
        param($Repo, $Cutoff)
        $branches = Get-AzDevOpsBranch -Repository $Repo.Id
        $branches | Where-Object { $_.LastCommitDate -lt $Cutoff } | Select-Object Name, RepositoryId, LastCommitDate
    } -ArgumentList $repo, $cutoffDate
}

# 等待所有作业完成并收集结果
$staleBranches = $jobs | Wait-Job | Receive-Job
# 清理作业
Remove-Job $jobs

方法三：使用Azure DevOps CLI批量查询

利用az devops CLI的批量能力，结合PowerShell的并行处理：

$orgUrl = "https://dev.azure.com/your-org"
$projectName = "your-project"
az devops configure --defaults organization=$orgUrl project=$projectName

$repos = az repos list --output json | ConvertFrom-Json
$staleBranches = @()

# 并行处理每个仓库
$repos | ForEach-Object -Parallel {
    $cutoff = (Get-Date).AddMonths(-3).ToString("yyyy-MM-ddTHH:mm:ssZ")
    # 通过CLI查询陈旧分支（利用--last-updated选项过滤）
    $stale = az repos list --repository $_.id --output json --last-updated "le $cutoff" | ConvertFrom-Json
    $stale | Select-Object name, @{Name="RepoId"; Expression={$_.repository.id}}, @{Name="LastCommitDate"; Expression={[DateTime]$_.lastCommit.date}}
} -ThrottleLimit 6 | ForEach-Object { $staleBranches += $_ }

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