首页 > 代码库 > AsyncEnumerableExtensions.cs z
AsyncEnumerableExtensions.cs z
public static class Extensions
{
public static async Task ForEachAsync<T, U>(this IEnumerable<T> collection, Func<T, Task<U>> body, IObserver<U> observer = null)
{
foreach (var item in collection)
{
var res = await body(item);
if (null != observer)
{
observer.OnNext(res);
}
}
}
public static async Task ForEachAsyncPerformance<T, U>(this IEnumerable<T> collection, Func<T, Task<U>> body, IObserver<U> observer = null)
{
var enumerator = collection.GetEnumerator();
Task<U> task = null;
if (enumerator.MoveNext())
{
task = body(enumerator.Current);
}
while (enumerator.MoveNext())
{
var item = enumerator.Current;
var res = await task;
task = body(item);
if (null != observer)
{
observer.OnNext(res);
}
}
if (null != task)
{
var res = await task;
if (null != observer)
{
observer.OnNext(res);
}
}
}
public static async Task ForEachAsync<T, U>(this IEnumerable<T> collection, int parallelism, Func<T, Task<U>> body, IObserver<U> observer = null)
{
int inFlight = 0;
var tasks = new HashSet<Task<U>>();
foreach (var item in collection)
{
if (inFlight >= parallelism)
{
var task = await Task.WhenAny(tasks);
tasks.Remove(task);
inFlight--;
if (null != observer)
{
observer.OnNext(task.Result);
}
}
inFlight++;
tasks.Add(body(item));
}
while (inFlight > 0)
{
var task = await Task.WhenAny(tasks);
tasks.Remove(task);
inFlight--;
if (null != observer)
{
observer.OnNext(task.Result);
}
}
}
}
声明:以上内容来自用户投稿及互联网公开渠道收集整理发布,本网站不拥有所有权,未作人工编辑处理,也不承担相关法律责任,若内容有误或涉及侵权可进行投诉: 投诉/举报 工作人员会在5个工作日内联系你,一经查实,本站将立刻删除涉嫌侵权内容。