c# - 带有外部参数的 Task.Run 循环由循环更改

Question

我有这个代码(简化)来处理 100 个不同并行线程上的参数数组，但是变量 x 和 y 在线程中使用时被线程内的循环更改。如果我用 1 个线程运行该函数，那么它就可以工作。

我还尝试将参数放在 ConcurrentBag 中并使用 foreach 进行循环，但结果相同，参数在线程中混合。

List<Task> tasks = new List<Task>();
var listConcurentBag = new ConcurrentBag<int>();
int nThreadCount = 0;

for (x=0; x<1000; x++)
  for (y=0; y<1000; y++)
  {
     int x1=x;
     int y2=y;

     Task t = Task.Run(() =>
     {         
        int param1=x1;
        int param2=y2;

        // some calculations with param1 and param2

        listConcurentBag.Add(result);
     }); // tasks

     tasks.Add(t);
     nThreadCount++;

   if (nThreadCount == 100) // after 100 threads started, wait
   {
       nThreadCount = 0;
       Task.WaitAll(tasks.ToArray());
       tasks.Clear();
   }
 }

Answer 1

您应该使用 Microsoft 的 Reactive Framework(又名 Rx)- NuGet System.Reactive 并添加 using System.Reactive.Linq; - 然后您可以这样做:

var query =
    from x in Observable.Range(0, 1000)
    from y in Observable.Range(0, 1000)
    from r in Observable.Start(() => GetResult(x,  y))
    select new { x, y, r };

IDisposable subscription =
    query
        .Buffer(100)
        .Subscribe(results =>
        {
            /* do something with each buffered list of results */
        });

现在这并不严格地与您当前的代码相同，但它会在使用线程池的最大容量可用时立即为您提供 100 个结果 block 。

您可以像这样更改它以设置并发性:

var query =
    from x in Observable.Range(0, 1000)
    from y in Observable.Range(0, 1000)
    select Observable.Start(() => new { x, y, r = GetResult(x,  y) });

IDisposable subscription =
    query
        .Merge(maxConcurrent: 100) // limit to 100 threads
        .Buffer(count: 100) // produce 100 results at a time
        .Subscribe(results =>
        {
            /* do something with the list of results */
        });

如果您想在代码自然完成之前停止代码，只需调用 subscription.Dispose();。

恕我直言，Rx 确实倾向于生成更简洁的代码。