동시 비동기 I / O 작업의 양을 제한하는 방법은 무엇입니까?
// let's say there is a list of 1000+ URLs
string[] urls = { "http://google.com", "http://yahoo.com", ... };
// now let's send HTTP requests to each of these URLs in parallel
urls.AsParallel().ForAll(async (url) => {
var client = new HttpClient();
var html = await client.GetStringAsync(url);
});
여기에 문제가 있습니다. 1000 개 이상의 동시 웹 요청을 시작합니다. 이러한 비동기 http 요청의 동시 양을 제한하는 쉬운 방법이 있습니까? 따라서 한 번에 20 개 이상의 웹 페이지가 다운로드되지 않도록합니다. 가장 효율적인 방법으로 수행하는 방법은 무엇입니까?
.NET 4.5 베타를 사용하여 .NET 용 비동기 최신 버전에서 확실히이 작업을 수행 할 수 있습니다. 'usr'의 이전 게시물은 Stephen Toub가 작성한 좋은 기사를 가리 키지 만 덜 알려진 소식은 비동기 세마포가 실제로 .NET 4.5의 베타 릴리스에 포함되었다는 것입니다.
우리가 사랑하는 SemaphoreSlim클래스 (원래보다 성능이 뛰어 나기 때문에 사용해야 함)를 살펴보면 Semaphore, 이제 WaitAsync(...)예상되는 모든 인수 (시간 초과 간격, 취소 토큰, 모든 일반적인 스케줄링 친구)와 함께 일련의 오버로드를 자랑합니다 . )
Stephen은 또한 베타와 함께 출시 된 새로운 .NET 4.5 기능에 대한 최신 블로그 게시물을 작성했습니다. What 's New for Parallelism in .NET 4.5 Beta를 참조하십시오 .
마지막으로, 비동기 메서드 조절에 SemaphoreSlim을 사용하는 방법에 대한 몇 가지 샘플 코드가 있습니다.
public async Task MyOuterMethod()
{
// let's say there is a list of 1000+ URLs
var urls = { "http://google.com", "http://yahoo.com", ... };
// now let's send HTTP requests to each of these URLs in parallel
var allTasks = new List<Task>();
var throttler = new SemaphoreSlim(initialCount: 20);
foreach (var url in urls)
{
// do an async wait until we can schedule again
await throttler.WaitAsync();
// using Task.Run(...) to run the lambda in its own parallel
// flow on the threadpool
allTasks.Add(
Task.Run(async () =>
{
try
{
var client = new HttpClient();
var html = await client.GetStringAsync(url);
}
finally
{
throttler.Release();
}
}));
}
// won't get here until all urls have been put into tasks
await Task.WhenAll(allTasks);
// won't get here until all tasks have completed in some way
// (either success or exception)
}
마지막으로 TPL 기반 스케줄링을 사용하는 솔루션을 언급 할 가치가있을 것입니다. 아직 시작되지 않은 TPL에서 위임 바인딩 작업을 생성하고 사용자 지정 작업 스케줄러가 동시성을 제한하도록 허용 할 수 있습니다. 실제로 여기에 MSDN 샘플이 있습니다.
TaskScheduler를 참조하십시오 .
IEnumerable (즉, URL 문자열)이 있고 이들 각각에 대해 I / O 바운드 작업을 동시에 수행하고 (즉, 비동기 http 요청 만들기) 선택적으로 최대 동시 수를 설정하려는 경우 실시간 I / O 요청을 수행하는 방법은 다음과 같습니다. 이렇게하면 스레드 풀 등을 사용하지 않고이 메서드는 세마포어 슬림을 사용하여 하나의 요청이 완료되고 세마포어를 떠나고 다음 요청이 들어오는 슬라이딩 창 패턴과 유사한 최대 동시 I / O 요청을 제어합니다.
사용법 : await ForEachAsync (urlStrings, YourAsyncFunc, optionalMaxDegreeOfConcurrency);
public static Task ForEachAsync<TIn>(
IEnumerable<TIn> inputEnumerable,
Func<TIn, Task> asyncProcessor,
int? maxDegreeOfParallelism = null)
{
int maxAsyncThreadCount = maxDegreeOfParallelism ?? DefaultMaxDegreeOfParallelism;
SemaphoreSlim throttler = new SemaphoreSlim(maxAsyncThreadCount, maxAsyncThreadCount);
IEnumerable<Task> tasks = inputEnumerable.Select(async input =>
{
await throttler.WaitAsync().ConfigureAwait(false);
try
{
await asyncProcessor(input).ConfigureAwait(false);
}
finally
{
throttler.Release();
}
});
return Task.WhenAll(tasks);
}
불행히도 .NET Framework에는 병렬 비동기 작업을 조정하기위한 가장 중요한 결합자가 없습니다. 그런 것은 내장되어 있지 않습니다.
가장 존경받는 Stephen Toub가 만든 AsyncSemaphore 클래스를 살펴보십시오 . 원하는 것은 세마포어라고하며 비동기 버전이 필요합니다.
오류의 경우 많은 함정이 있고 세마포어를 직접 사용하는 것이 까다로울 수 있으므로 바퀴를 다시 발명하는 대신 AsyncEnumerator NuGet 패키지 를 사용하는 것이 좋습니다 .
// let's say there is a list of 1000+ URLs
string[] urls = { "http://google.com", "http://yahoo.com", ... };
// now let's send HTTP requests to each of these URLs in parallel
await urls.ParallelForEachAsync(async (url) => {
var client = new HttpClient();
var html = await client.GetStringAsync(url);
}, maxDegreeOfParalellism: 20);
Theo Yaung 예제는 좋지만 대기 작업 목록이없는 변형이 있습니다.
class SomeChecker
{
private const int ThreadCount=20;
private CountdownEvent _countdownEvent;
private SemaphoreSlim _throttler;
public Task Check(IList<string> urls)
{
_countdownEvent = new CountdownEvent(urls.Count);
_throttler = new SemaphoreSlim(ThreadCount);
return Task.Run( // prevent UI thread lock
async () =>{
foreach (var url in urls)
{
// do an async wait until we can schedule again
await _throttler.WaitAsync();
ProccessUrl(url); // NOT await
}
//instead of await Task.WhenAll(allTasks);
_countdownEvent.Wait();
});
}
private async Task ProccessUrl(string url)
{
try
{
var page = await new WebClient()
.DownloadStringTaskAsync(new Uri(url));
ProccessResult(page);
}
finally
{
_throttler.Release();
_countdownEvent.Signal();
}
}
private void ProccessResult(string page){/*....*/}
}
SemaphoreSlim은 여기에서 매우 유용 할 수 있습니다. 내가 만든 확장 방법은 다음과 같습니다.
/// <summary>
/// Concurrently Executes async actions for each item of <see cref="IEnumerable<typeparamref name="T"/>
/// </summary>
/// <typeparam name="T">Type of IEnumerable</typeparam>
/// <param name="enumerable">instance of <see cref="IEnumerable<typeparamref name="T"/>"/></param>
/// <param name="action">an async <see cref="Action" /> to execute</param>
/// <param name="maxActionsToRunInParallel">Optional, max numbers of the actions to run in parallel,
/// Must be grater than 0</param>
/// <returns>A Task representing an async operation</returns>
/// <exception cref="ArgumentOutOfRangeException">If the maxActionsToRunInParallel is less than 1</exception>
public static async Task ForEachAsyncConcurrent<T>(
this IEnumerable<T> enumerable,
Func<T, Task> action,
int? maxActionsToRunInParallel = null)
{
if (maxActionsToRunInParallel.HasValue)
{
using (var semaphoreSlim = new SemaphoreSlim(
maxActionsToRunInParallel.Value, maxActionsToRunInParallel.Value))
{
var tasksWithThrottler = new List<Task>();
foreach (var item in enumerable)
{
// Increment the number of currently running tasks and wait if they are more than limit.
await semaphoreSlim.WaitAsync();
tasksWithThrottler.Add(Task.Run(async () =>
{
await action(item).ContinueWith(res =>
{
// action is completed, so decrement the number of currently running tasks
semaphoreSlim.Release();
});
}));
}
// Wait for all of the provided tasks to complete.
await Task.WhenAll(tasksWithThrottler.ToArray());
}
}
else
{
await Task.WhenAll(enumerable.Select(item => action(item)));
}
}
샘플 사용법 :
await enumerable.ForEachAsyncConcurrent(
async item =>
{
await SomeAsyncMethod(item);
},
5);
1000 개의 작업이 매우 빠르게 대기열에 추가 될 수 있지만 Parallel Tasks 라이브러리는 컴퓨터의 CPU 코어 양과 동일한 동시 작업 만 처리 할 수 있습니다. 즉, 4 코어 시스템이있는 경우 주어진 시간에 4 개의 작업 만 실행됩니다 (MaxDegreeOfParallelism을 낮추지 않는 한).
CPU 바운드 작업의 속도를 높이려면 병렬 계산을 사용해야합니다. 여기서는 I / O 바운드 작업에 대해 설명합니다. 다중 코어 CPU에서 바쁜 단일 코어를 압도하지 않는 한 구현은 순전히 비동기 이어야합니다 .
편집 나는 여기에 "비동기 세마포어"를 사용하는 usr의 제안을 좋아합니다.
MaxDegreeOfParallelism에서 지정할 수있는 옵션 인을 사용하십시오 Parallel.ForEach().
var options = new ParallelOptions { MaxDegreeOfParallelism = 20 };
Parallel.ForEach(urls, options,
url =>
{
var client = new HttpClient();
var html = client.GetStringAsync(url);
// do stuff with html
});
기본적으로 적중하려는 각 URL에 대해 작업 또는 작업을 만들고 목록에 넣은 다음 해당 목록을 처리하여 병렬로 처리 할 수있는 수를 제한하려고합니다.
내 블로그 게시물 은 Tasks와 Actions를 사용하여이 작업을 수행하는 방법을 보여주고, 다운로드하여 실행할 수있는 샘플 프로젝트를 제공합니다.
액션
Actions를 사용하는 경우 내장 된 .Net Parallel.Invoke 함수를 사용할 수 있습니다. 여기서는 최대 20 개의 스레드를 병렬로 실행하도록 제한합니다.
var listOfActions = new List<Action>();
foreach (var url in urls)
{
var localUrl = url;
// Note that we create the Task here, but do not start it.
listOfTasks.Add(new Task(() => CallUrl(localUrl)));
}
var options = new ParallelOptions {MaxDegreeOfParallelism = 20};
Parallel.Invoke(options, listOfActions.ToArray());
작업 포함
Tasks에는 기본 제공 기능이 없습니다. 그러나 내 블로그에서 제공하는 것을 사용할 수 있습니다.
/// <summary>
/// Starts the given tasks and waits for them to complete. This will run, at most, the specified number of tasks in parallel.
/// <para>NOTE: If one of the given tasks has already been started, an exception will be thrown.</para>
/// </summary>
/// <param name="tasksToRun">The tasks to run.</param>
/// <param name="maxTasksToRunInParallel">The maximum number of tasks to run in parallel.</param>
/// <param name="cancellationToken">The cancellation token.</param>
public static async Task StartAndWaitAllThrottledAsync(IEnumerable<Task> tasksToRun, int maxTasksToRunInParallel, CancellationToken cancellationToken = new CancellationToken())
{
await StartAndWaitAllThrottledAsync(tasksToRun, maxTasksToRunInParallel, -1, cancellationToken);
}
/// <summary>
/// Starts the given tasks and waits for them to complete. This will run the specified number of tasks in parallel.
/// <para>NOTE: If a timeout is reached before the Task completes, another Task may be started, potentially running more than the specified maximum allowed.</para>
/// <para>NOTE: If one of the given tasks has already been started, an exception will be thrown.</para>
/// </summary>
/// <param name="tasksToRun">The tasks to run.</param>
/// <param name="maxTasksToRunInParallel">The maximum number of tasks to run in parallel.</param>
/// <param name="timeoutInMilliseconds">The maximum milliseconds we should allow the max tasks to run in parallel before allowing another task to start. Specify -1 to wait indefinitely.</param>
/// <param name="cancellationToken">The cancellation token.</param>
public static async Task StartAndWaitAllThrottledAsync(IEnumerable<Task> tasksToRun, int maxTasksToRunInParallel, int timeoutInMilliseconds, CancellationToken cancellationToken = new CancellationToken())
{
// Convert to a list of tasks so that we don't enumerate over it multiple times needlessly.
var tasks = tasksToRun.ToList();
using (var throttler = new SemaphoreSlim(maxTasksToRunInParallel))
{
var postTaskTasks = new List<Task>();
// Have each task notify the throttler when it completes so that it decrements the number of tasks currently running.
tasks.ForEach(t => postTaskTasks.Add(t.ContinueWith(tsk => throttler.Release())));
// Start running each task.
foreach (var task in tasks)
{
// Increment the number of tasks currently running and wait if too many are running.
await throttler.WaitAsync(timeoutInMilliseconds, cancellationToken);
cancellationToken.ThrowIfCancellationRequested();
task.Start();
}
// Wait for all of the provided tasks to complete.
// We wait on the list of "post" tasks instead of the original tasks, otherwise there is a potential race condition where the throttler's using block is exited before some Tasks have had their "post" action completed, which references the throttler, resulting in an exception due to accessing a disposed object.
await Task.WhenAll(postTaskTasks.ToArray());
}
}
그런 다음 작업 목록을 만들고 한 번에 최대 20 개를 동시에 실행하도록 함수를 호출하면 다음과 같이 할 수 있습니다.
var listOfTasks = new List<Task>();
foreach (var url in urls)
{
var localUrl = url;
// Note that we create the Task here, but do not start it.
listOfTasks.Add(new Task(async () => await CallUrl(localUrl)));
}
await Tasks.StartAndWaitAllThrottledAsync(listOfTasks, 20);
Old question, new answer. @vitidev had a block of code that was reused almost intact in a project I reviewed. After discussing with a few colleagues one asked "Why don't you just use the built-in TPL methods?" ActionBlock looks like the winner there. https://msdn.microsoft.com/en-us/library/hh194773(v=vs.110).aspx. Probably won't end up changing any existing code but will definitely look to adopt this nuget and reuse Mr. Softy's best practice for throttled parallelism.
Here is a solution that takes advantage of the lazy nature of LINQ. It offers the advantage of not spawning threads (like the accepted answer does), and not having all tasks created at once and almost all of them blocked on a SemaphoreSlim, like the SemaphoreSlim solutions. At first lets make it work without throttling. The first step is to convert our urls to an enumerable of tasks.
string[] urls =
{
"https://stackoverflow.com",
"https://superuser.com",
"https://serverfault.com",
"https://meta.stackexchange.com",
// ...
};
var httpClient = new HttpClient();
var tasks = urls.Select(async (url) =>
{
return (Url: url, Html: await httpClient.GetStringAsync(url));
});
The second step is to await all tasks concurrently using the Task.WhenAll method:
var results = await Task.WhenAll(tasks);
foreach (var result in results)
{
Console.WriteLine($"Url: {result.Url}, {result.Html.Length:#,0} chars");
}
Output:
Url: https://stackoverflow.com, 105.574 chars
Url: https://superuser.com, 126.953 chars
Url: https://serverfault.com, 125.963 chars
Url: https://meta.stackexchange.com, 185.276 chars
...
Microsoft's implementation of Task.WhenAll materializes instantly the supplied enumerable to an array, causing all tasks to starts at once. We don't want that, because we want to limit the number of concurrent asynchronous operations. So we'll need to implement an alternative WhenAll that will enumerate our enumerable gently and slowly. We will do that by creating a number of worker-tasks (equal to the desired degree of parallelism), and each worker-task will enumerate our enumerable one task at a time, using a lock to ensure that each url-task will be processed by only one worker-task. Then we await for all worker-tasks to complete, and finally we return the results after we restore their order. Here is the implementation:
public static async Task<T[]> WhenAll<T>(IEnumerable<Task<T>> tasks,
int degreeOfParallelism)
{
if (tasks is ICollection<Task<T>>) throw new ArgumentException(
"The enumerable should not be materialized.", nameof(tasks));
var results = new List<(int Index, T Result)>();
var failed = false;
using (var enumerator = tasks.GetEnumerator())
{
int index = 0;
var workerTasks = Enumerable.Range(0, degreeOfParallelism)
.Select(async _ =>
{
try
{
while (true)
{
Task<T> task;
int localIndex;
lock (enumerator)
{
if (failed || !enumerator.MoveNext()) break;
task = enumerator.Current;
localIndex = index++;
}
var result = await task.ConfigureAwait(false);
lock (results) results.Add((localIndex, result));
}
}
catch
{
lock (enumerator) failed = true;
throw;
}
}).ToArray();
await Task.WhenAll(workerTasks).ConfigureAwait(false);
}
return results.OrderBy(e => e.Index).Select(e => e.Result).ToArray();
}
...and here is what we must change in our initial code, to achieve the desired throttling:
var results = await WhenAll(tasks, degreeOfParallelism: 2);
There is a difference regarding the handling of the exceptions. The native Task.WhenAll waits for all tasks to complete and aggregates all the exceptions. The implementation above stops waiting shortly after the completion of the first faulted task.
this isn't a general solution for async but, for HttpClient you can simply try
System.Net.ServicePointManager.DefaultConnectionLimit = 20;
'Programming' 카테고리의 다른 글
| 다른 호스트의 하위 도메인 (0) | 2020.08.26 |
|---|---|
| exec 결과를 SQL 변수에 할당하는 방법은 무엇입니까? (0) | 2020.08.26 |
| SAML과 OAuth를 사용한 페더레이션 로그인 (0) | 2020.08.26 |
| 다른 작업을 수행하는 동안 커밋되지 않은 변경 사항을 제쳐두고 어떻게해야합니까? (0) | 2020.08.26 |
| HTML / CSS를 사용하여 이미지 주위에 텍스트를 감싸는 방법 (0) | 2020.08.26 |