【问题标题】:Restricting the enumerations of LINQ queries to One Only将 LINQ 查询的枚举限制为仅一个
【发布时间】:2019-08-29 03:36:39
【问题描述】:

我有一个不应多次枚举的 LINQ 查询,我想避免错误地枚举它两次。我可以使用任何扩展方法来确保我免受此类错误的影响吗?我正在考虑这样的事情:

var numbers = Enumerable.Range(1, 10).OnlyOnce();
Console.WriteLine(numbers.Count()); // shows 10
Console.WriteLine(numbers.Count()); // throws InvalidOperationException: The query cannot be enumerated more than once.

我想要这个功能的原因是因为我有一个可枚举的任务,旨在逐步实例化和运行任务,同时在控制下缓慢枚举。我已经犯了两次运行任务的错误,因为我忘记了它是一个不同的可枚举而不是 一个数组。

var tasks = Enumerable.Range(1, 10).Select(n => Task.Run(() => Console.WriteLine(n)));
Task.WaitAll(tasks.ToArray()); // Lets wait for the tasks to finish...
Console.WriteLine(String.Join(", ", tasks.Select(t => t.Id))); // Lets see the completed task IDs...
// Oups! A new set of tasks started running!

【问题讨论】:

  • 引入自定义类,该类将在后台使用 Queue。实现GetEnumerator,它将在每次迭代时从队列中删除项目——这样你就可以安全地迭代你的类,而无需多次执行任务,因为任务将在第一次迭代时被删除。使用ImmutableQueue 获取线程安全类。

标签: c# linq task-parallel-library


【解决方案1】:

我想避免错误地枚举它两次。

您可以使用一个集合包装该集合,如果它被枚举两次则抛出该集合。

例如:

using System;
using System.Collections;
using System.Collections.Generic;
using System.Linq;

namespace ConsoleApp8
{
    public static class EnumExtension
    {
        class OnceEnumerable<T> : IEnumerable<T>
        {
            IEnumerable<T> col;
            bool hasBeenEnumerated = false;
            public OnceEnumerable(IEnumerable<T> col)
            {
                this.col = col;
            }

            public IEnumerator<T> GetEnumerator()
            {
                if (hasBeenEnumerated)
                {
                    throw new InvalidOperationException("This collection has already been enumerated.");
                }
                this.hasBeenEnumerated = true;
                return col.GetEnumerator();
            }

            IEnumerator IEnumerable.GetEnumerator()
            {
                return GetEnumerator();
            }
        }

        public static IEnumerable<T> OnlyOnce<T>(this IEnumerable<T> col)
        {
            return new OnceEnumerable<T>(col);
        }
    }
    class Program
    {
        static void Main(string[] args)
        {
             var col = Enumerable.Range(1, 10).OnlyOnce();

             var colCount = col.Count(); //first enumeration
             foreach (var c in col) //second enumeration
             {
                 Console.WriteLine(c);
             }
        }
    }
}

【讨论】:

  • 不错的答案,赞成,但是我觉得我们陷入了 xy 问题
  • @Fabio 我知道如何在需要时使其线程安全。但目前我的可枚举用于单个执行流程,因此我将按原样使用 David 的解决方案。
  • @Michael Randall 这是一个 XY 问题,但 X 比 Y 宽。因为我的任务问题可以通过使用更强大的实现来解决,但我最初请求的可枚举可以是仅枚举一次也可以有其他应用程序。很高兴在我的工具箱中有 David Browne 的 OnceEnumerable 课程。 :-)
【解决方案2】:

Rx 当然是控制并行性的一个选项。

var query =
    Observable
        .Range(1, 10)
        .Select(n => Observable.FromAsync(() => Task.Run(() => new { Id = n })));

var tasks = query.Merge(maxConcurrent: 3).ToArray().Wait();

Console.WriteLine(String.Join(", ", tasks.Select(t => t.Id)));

【讨论】:

  • 谢谢,很有趣。我尝试了 sn-p,它实际上做了应该做的事情。
  • Rx 真的很棒。我将它用于我的所有事件处理和处理任务时。它更强大,恕我直言。
【解决方案3】:

枚举枚举,故事结束。您只需拨打ToListToArray

// this will enumerate and start the tasks
var tasks = Enumerable.Range(1, 10)
                      .Select(n => Task.Run(() => Console.WriteLine(n)))
                      .ToList();

// wait for them all to finish
Task.WaitAll(tasks.ToArray());
Console.WriteLine(String.Join(", ", tasks.Select(t => t.Id)));

如果你想要并行性,Hrm

Parallel.For(0, 100, index => Console.WriteLine(index) );

或者如果您使用的是异步和等待模式

public static async Task DoWorkLoads(IEnumerable <Something> results)
{
   var options = new ExecutionDataflowBlockOptions
                     {
                        MaxDegreeOfParallelism = 50
                     };

   var block = new ActionBlock<Something>(MyMethodAsync, options);

   foreach (var result in results)
      block.Post(result);

   block.Complete();
   await block.Completion;

}

...

public async Task MyMethodAsync(Something result)
{       
   await SomethingAsync(result);
}

更新,既然你正在寻找一种控制最大并发度的方法,你可以使用这个

public static async Task<IEnumerable<Task>> ExecuteInParallel<T>(this IEnumerable<T> collection,Func<T, Task> callback,int degreeOfParallelism)
{
   var queue = new ConcurrentQueue<T>(collection);

   var tasks = Enumerable.Range(0, degreeOfParallelism)
                         .Select(async _ =>
                          {
                             while (queue.TryDequeue(out var item))
                                await callback(item);
                          })
                         .ToArray();

   await Task.WhenAll(tasks);

   return tasks;
}

【讨论】:

  • 我不想打电话给ToArray,因为那样所有任务都会立即开始运行。我想慢慢枚举任务(以实现最大程度的并行性)。
  • @TheodorZoulias 然后使用 ActionBlock 或响应式扩展
  • 这些都有学习曲线。在我研究它们之后,我肯定会使用它们。目前我对 Task Parallel Library 非常了解,并且我更希望能够以我目前的知识水平立即应用的解决方案。
  • @TheodorZoulias 请解释这是什么意思I want to enumerate the tasks slowly
猜你喜欢
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 2017-03-30
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 2012-08-10
相关资源
最近更新 更多