c# - 如何为 PLINQ 编写线程感知扩展函数？

Question

有人知道如何在 PLINQ 中编写返回 ParallelQuery 的扩展函数吗？

更具体地说，我有以下问题:我想在需要引擎的 PLINQ 查询中执行转换，引擎的创建成本高且无法同时访问。

我可以做以下事情:

var result = source.AsParallel ().Select ( (i) => { var e = new Engine (); return e.Process(i); } )

这里，引擎为每个项目创建一次，这太昂贵了。

我希望每个线程创建一次引擎。

有了聚合，我可以用类似的东西接近我想要的东西

// helper class: engine to use plus list of results obtained in thread so far
class EngineAndResults {
   public Engine engine = null;
   public IEnumerable<ResultType> results;
}

var result = source.AsParallel ().Aggregate (

   // done once per block of items (=thread),
   // returning an empty list, but a new engine
   () => new EngineAndList () {
       engine = new Engine (),
       results = Enumerable.Empty<ResultType> ()
   },

   // we process a new item and put it to the thread-local list,
   // preserving the engine for further use
   (engineAndResults, item) => new EngineAndResults () {
       engine = engineAndResults.engine,
       results = Enumerable.Concat (
           engineAndResults.results,
           new ResultType [] { engineAndResults.engine.Process (item) }
       )
   },

   // tell linq how to aggregate across threads
   (engineAndResults1, engineAndResults2) => new EngineAndResults () {
       engine = engineAndResults1.engine,
       results = Enumerable.Concat (engineAndResults1.results, engineAndResults2.results)
   },

   // after all aggregations, how do we come to the result?
   engineAndResults => engineAndResults.results
);

如您所见，我滥用累加器为每个线程携带一个引擎。这里的问题是 PLINQ 最后将结果聚合到单个 IEnumerable 中，这会导致线程同步。如果我想在之后附加另一个 PLINQ 扩展，这不是很好。

我会喜欢这样的东西

   var result = source.AsParallel ()
                  .SelectWithThreadwiseInitWhichIAmLookingFor (
                       () => new Engine (),
                       (engine, item) => engine.Process (item)
              )

有人知道如何实现吗？

Answer 1

你可以使用 ThreadLocal<T> 去做这个。像这样的东西:

var engine = new ThreadLocal<Engine>(() => new Engine());
var result = source.AsParallel()
                   .Select(item => engine.Value.Process(item));