使用Parallel.Foreach结合OledbDataReader调用Web API出现重复行问题

我之前也踩过Parallel.Foreach处理API调用+数据库插入时结果和普通循环不一致的坑，结合你说的10行地址数据、API返回标准化地址但结果异常的场景，大概率是线程安全问题或者并行调用带来的资源冲突导致的，咱们一步步拆解：

常见原因分析 #

1. 共享对象/变量未做线程隔离 #

如果你在循环里复用了同一个实体对象、DbContext或者API请求容器，多个线程会同时读写这些共享资源，直接导致数据覆盖或者混乱。比如这种错误写法：

// 错误：多个线程共享同一个response对象和DbContext
var sharedResponse = new NormalizedAddress();
using (var sharedDbContext = new MyDbContext())
{
    Parallel.ForEach(dataRows, row => {
        sharedResponse = apiClient.CallAddressApi(row.Address).Result;
        sharedDbContext.NormalizedAddresses.Add(sharedResponse);
        sharedDbContext.SaveChanges();
    });
}

这里不同线程会同时修改sharedResponse，而且EF的DbContext本身就不是线程安全的，并行操作会直接搞乱上下文的状态。

2. API速率限制/并发限制触发 #

有些地址标准化API会限制并发请求数或者每秒请求量，并行调用时如果超过限制，API可能会返回缓存结果、错误响应或者默认值，导致你看到的结果和单线程调用不一致。

3. 未控制并行度导致的资源竞争 #

默认Parallel.Foreach会根据CPU核心数设置并行度，如果并行数太高，不仅会触发API限制，还可能导致本地网络、数据库连接池的资源竞争，间接引发数据异常。

对应解决方案 #

1. 为每个迭代创建独立资源 #

确保每个循环线程都使用自己的API请求实例、响应对象和数据库上下文：

// 正确写法：每个迭代隔离资源
Parallel.ForEach(dataRows, new ParallelOptions { MaxDegreeOfParallelism = 3 }, row => {
    // 每个线程单独创建DbContext
    using (var dbContext = new MyDbContext())
    {
        // 每个请求单独创建响应对象
        var apiResponse = apiClient.CallAddressApi(row.Address).Result;
        var normalizedAddr = apiResponse.Content.ReadFromJsonAsync<NormalizedAddress>().Result;
        
        // 插入数据库，每个上下文独立提交
        dbContext.NormalizedAddresses.Add(normalizedAddr);
        dbContext.SaveChanges();
    }
});

2. 控制并行度 #

通过ParallelOptions.MaxDegreeOfParallelism设置合理的并行数，比如根据API的并发限制设置为2-5，避免触发限流：

var parallelOptions = new ParallelOptions {
    MaxDegreeOfParallelism = 2 // 匹配API的并发限制
};
Parallel.ForEach(dataRows, parallelOptions, row => { /* ... */ });

3. 增加日志排查 #

在每个迭代里记录输入地址、API返回结果和线程ID，方便定位哪一步出了问题：

Parallel.ForEach(dataRows, row => {
    var threadId = Thread.CurrentThread.ManagedThreadId;
    Console.WriteLine($"线程{threadId}处理地址：{row.Address}");
    
    var apiResponse = apiClient.CallAddressApi(row.Address).Result;
    var normalizedAddr = apiResponse.Content.ReadFromJsonAsync<NormalizedAddress>().Result;
    
    Console.WriteLine($"线程{threadId}返回结果：{normalizedAddr.FullAddress}");
    // 插入数据库...
});

额外排查建议 #

• 先拿2行测试数据对比并行和单线程的结果，看是不是特定数据触发了问题
• 检查API文档的并发限制条款，确认并行数是否合规
• 如果用的是HttpClient，确保是全局单例（HttpClient本身线程安全，但不要每次迭代都新建，会耗尽端口）

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