Parallel.ForEach重复处理同一S3Object问题求助

你的问题根本原因不是Parallel.ForEach重复迭代了同一个fileObject，而是代码里的fileRequest是多个线程共享的实例，存在竞态条件：
当多个线程同时修改fileRequest.Key时，比如线程A刚把Key设为文件X，还没执行GetObjectAsync，线程B就把Key改成了文件Y，或者反过来，最终导致两个线程可能拿到同一个Key的请求，出现重复处理的情况。

以下是几种解决/规避方案：

1. 每个线程使用独立的请求对象（推荐） #

彻底避免共享变量的竞态问题，在循环内部创建GetObjectRequest实例：

List<S3Object> fileObjects = listFilesResponse.Result.S3Objects;
filecount += listFilesResponse.Result.KeyCount;

ParallelOptions paralleOpts = new ParallelOptions
{
    MaxDegreeOfParallelism = Environment.ProcessorCount
};

Parallel.ForEach(fileObjects, paralleOpts, fileObject =>
{
    // 每个线程创建独立的请求对象，避免共享
    var fileRequest = new GetObjectRequest
    {
        Key = fileObject.Key,
        BucketName = "你的Bucket名称" // 补充必要参数
    };

    using (var fileResponse = s3Client.GetObjectAsync(fileRequest).Result)
    using (Stream responseStream = fileResponse.ResponseStream)
    using (StreamReader reader = new StreamReader(responseStream))
    {
        //do stuff with file...
    }
});

2. 对共享请求对象加锁保护（不推荐，会降低并行效率） #

如果必须复用fileRequest实例，需要用锁把修改Key和执行请求的过程保护起来，确保同一时间只有一个线程操作：

List<S3Object> fileObjects = listFilesResponse.Result.S3Objects;
filecount += listFilesResponse.Result.KeyCount;

ParallelOptions paralleOpts = new ParallelOptions
{
    MaxDegreeOfParallelism = Environment.ProcessorCount
};
object requestLock = new object();

Parallel.ForEach(fileObjects, paralleOpts, fileObject =>
{
    GetObjectResponse fileResponse;
    lock (requestLock)
    {
        fileRequest.Key = fileObject.Key;
        // 必须在锁内等待异步操作完成，否则竞态问题依然存在
        fileResponse = s3Client.GetObjectAsync(fileRequest).Result;
    }

    using (fileResponse)
    using (Stream responseStream = fileResponse.ResponseStream)
    using (StreamReader reader = new StreamReader(responseStream))
    {
        //do stuff with file...
    }
});

3. 改用异步并行处理（更适合I/O密集型操作） #

S3文件读取属于I/O密集型任务，Parallel.ForEach更适合CPU密集型场景，改用Task.WhenAll的异步并行方式更高效，同时天然避免共享变量问题：

List<S3Object> fileObjects = listFilesResponse.Result.S3Objects;
filecount += listFilesResponse.Result.KeyCount;

// 生成所有异步任务
var tasks = fileObjects.Select(async fileObject =>
{
    var fileRequest = new GetObjectRequest
    {
        Key = fileObject.Key,
        BucketName = "你的Bucket名称" // 补充必要参数
    };

    using (var fileResponse = await s3Client.GetObjectAsync(fileRequest))
    using (Stream responseStream = fileResponse.ResponseStream)
    using (StreamReader reader = new StreamReader(responseStream))
    {
        //do stuff with file...
    }
});

// 等待所有任务完成
await Task.WhenAll(tasks);

另外补充：你之前尝试用List<string>或ConcurrentBag<string>记录已处理文件的方法无效，是因为问题根源不在迭代重复，而是共享请求对象导致实际处理的Key和当前迭代的fileObject不一致，所以记录已处理文件无法解决本质问题。

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