这个线程代码有解释吗？

Question

所以遇到了一些与此非常相似的代码。我只是想知道是否有人可以向我解释一下。

看看它如何使用 RX 调度程序，然后是 Parallel.For 以及在其中一个新的 TaskFactory.StartNew

    IDisposable subscription = someObservable.ObserveOn(ThreadPoolScheduler.Instance)
    .Subscribe(o =>
       {
           Parallel.ForEach(xxxs,
               x =>
               {
                   var theKey = x.Key;
                   if (!theTasks.ContainsKey(theKey) ||
                       theTasks.ContainsKey(theKey) && theTasks[theKey].IsCompleted)
                   {
                       theTasks[theKey] = Task.Factory.StartNew(
                           () =>
                           {
                               .....
                               }
                               catch (CommunicationObjectAbortedException ex)
                               {
                               ....
                               }
                               catch (ObjectDisposedException ex)
                               {
                               ....
                               }
                               catch (Exception e)
                               {
                               ....
                               }
                           });
                   }
               });
       },
       ex =>
       {
          ....
       },
       () =>
       {
          ....
       });
    } 

我知道所有这些事情单独做了什么，但我不确定这里的组合线程效果是什么。谁能猜一猜

Answer 1

啊，是的，并发Turducken。

ThreadPoolScheduler 将工作安排在与 任务池 不同的 线程池 上。 ThreadPoolScheduler 用于在没有任务池的平台上使用 - 如果可能，更喜欢 TaskPoolScheduler。

感觉就像作者试图通过使用线程池来仅为手头的任务（请原谅双关语）保存任务池。

Parallel.ForEach 阻塞，直到循环完成。因此，当它在线程池上运行时，当发出新项目时，请在线程池中借用的线程上执行下一个ForEach。

至于内部位，作者希望每个唯一键运行一个 Task，如果尚未运行。