我有一個foreach回圈，負責執行一組特定的陳述句。其中一部分是將影像從 URL 保存到 Azure 存盤。我必須為大量資料執行此操作。為了達到同樣的效果，我將foreach回圈轉換為Parallel.ForEach回圈。
Parallel.ForEach(listSkills, item =>
{
// some business logic
var b = getImageFromUrl(item.Url);
Stream ms = new MemoryStream(b);
saveImage(ms);
// more business logic
});
private static byte[] getByteArray(Stream input)
{
using (MemoryStream ms = new MemoryStream())
{
input.CopyTo(ms);
return ms.ToArray();
}
}
public static byte[] getImageFromUrl(string url)
{
HttpWebRequest request = null;
HttpWebResponse response = null;
byte[] b = null;
request = (HttpWebRequest)WebRequest.Create(url);
response = (HttpWebResponse)request.GetResponse();
if (request.HaveResponse)
{
if (response.StatusCode == HttpStatusCode.OK)
{
Stream receiveStream = response.GetResponseStream();
b = getByteArray(receiveStream);
}
}
return b;
}
public static void saveImage(Stream fileContent)
{
fileContent.Seek(0, SeekOrigin.Begin);
byte[] bytes = getByteArray(fileContent);
var blob = null;
blob.UploadFromByteArrayAsync(bytes, 0, bytes.Length).Wait();
}
盡管在某些情況下我收到以下錯誤并且影像未保存。
現有連接被遠程主機強行關閉。
還分享 StackTrace ：
at System.Net.Sockets.NetworkStream.Read(Span`1 buffer)
at System.Net.Security.SslStream.<FillBufferAsync>d__183`1.MoveNext()
at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
at System.Net.Security.SslStream.<ReadAsyncInternal>d__181`1.MoveNext()
at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
at System.Net.Security.SslStream.Read(Byte[] buffer, Int32 offset, Int32 count)
at System.IO.Stream.Read(Span`1 buffer)
at System.Net.Http.HttpConnection.Read(Span`1 destination)
at System.Net.Http.HttpConnection.ContentLengthReadStream.Read(Span`1 buffer)
at System.Net.Http.HttpBaseStream.Read(Byte[] buffer, Int32 offset, Int32 count)
at System.IO.Stream.CopyTo(Stream destination, Int32 bufferSize)
at Utilities.getByteArray(Stream input) in D:\repos\SampleProj\Sample\Helpers\CH.cs:line 238
at Utilities.getImageFromUrl(String url) in D:\repos\SampleProj\Sample\Helpers\CH.cs:line 178
我猜這可能是因為我沒有使用鎖？我不確定是否在Parallel.ForEach回圈中使用鎖。
uj5u.com熱心網友回復：
根據關于 stackoverflow 的另一個問題，這里是現有連接被遠程主機強行關閉的潛在原因。：
- 您正在向應用程式發送格式錯誤的資料（可能包括向 HTTP 服務器發送 HTTPS 請求）
- 由于某種原因，客戶端和服務器之間的網路鏈接正在關閉
- 您在第三方應用程式中觸發了導致其崩潰的錯誤
- 第三方應用程式已耗盡系統資源
由于只有您的一些請求受到影響，我認為我們可以排除第一個請求。當然，這可能是網路問題，在這種情況下，這會不時發生，具體取決于您和服務器之間的網路質量。
除非您從其他用戶那里找到 AzureStorage 錯誤的跡象，否則您的呼叫很可能同時消耗了太多遠程服務器的資源（連接/資料）。服務器和代理對它們可以同時處理的連接數量有限制（尤其是來自同一臺客戶端機器）。
根據listSkills串列的大小，您的代碼可能會并行啟動大量請求（與您的執行緒池一樣多），可能會淹沒服務器。
您至少可以使用以下方式限制并行任務啟動的數量MaxDegreeOfParallelism：
Parallel.ForEach(listSkills,
new ParallelOptions { MaxDegreeOfParallelism = 4 },
item =>
{
// some business logic
var b = getImageFromUrl(item.Url);
Stream ms = new MemoryStream(b);
saveImage(ms);
// more business logic
});
uj5u.com熱心網友回復：
您可以控制并行性，例如：
listSkills.AsParallel()
.Select(item => {/*Your Logic*/ return item})
.WithDegreeOfParallelism(10)
.Select(item =>
{
getImageFromUrl(item.url);
saveImage(your_stream);
return item;
});
但Parallel.ForEach不好，IO因為它是為CPU-intensive任務設計的，如果你將它用于IO-bound專門發出 Web 請求的操作，你可能會浪費執行緒池執行緒在等待回應時阻塞。
您使用異步 Web 請求方法，例如HttpWebRequest.GetResponseAsync，另一方面，您也可以為此使用執行緒同步結構，例如使用Semaphore，Semaphore就像佇列一樣，它允許X執行緒通過，其余的應該等到一個繁忙的執行緒完成是作業。首先使您的getStream方法像async這樣（這不是好的解決方案，但可以更好）：
public static async Task getImageFromUrl(SemaphoreSlim semaphore, string url)
{
try
{
HttpWebRequest request = null;
byte[] b = null;
request = (HttpWebRequest)WebRequest.Create(url);
using (var response = await request.GetResponseAsync().ConfigureAwait(false))
{
// your logic
}
}
catch (Exception ex)
{
// handle exp
}
finally
{
// release
semaphore.Release();
}
}
然后：
using (var semaphore = new SemaphoreSlim(10))
{
foreach (var url in urls)
{
// await here until there is a room for this task
await semaphore.WaitAsync();
tasks.Add(getImageFromUrl(semaphore, url));
}
// await for the rest of tasks to complete
await Task.WhenAll(tasks);
}
您不應該使用，Parallel或者Task.Run您可以使用以下async處理程式方法：
public async Task handleResponse(Task<HttpResponseMessage> response)
{
HttpResponseMessage response = await response;
//Process your data
}
然后Task.WhenAll像這樣使用：
Task[] requests = myList.Select(l => getImageFromUrl(l.Id))
.Select(r => handleResponse(r))
.ToArray();
await Task.WhenAll(requests);
最后，您的場景有幾種解決方案，但忘記Parallel.Foreach使用優化的解決方案。
uj5u.com熱心網友回復：
這段代碼有幾個問題：
Parallel.ForEach用于資料并行，而不是 IO。代碼凍結所有等待 IO 完成的 CPU 內核- HttpWebRequest 是 .NET Core 中 HttpClient 的包裝器。使用 HttpWebRequest 效率低下，而且比需要的復雜得多。
- HttpClient 可以發布檢索或發布流內容，這意味著沒有理由將流內容加載到記憶體中。HttpClient 是執行緒安全的，也可以重用。
在 .NET Core 中有多種方法可以同時執行許多 IO 操作。
.NET 6
在當前的 .NET 長期支持版本 .NET 6 中，這可以使用Parallel.ForEachAsync來完成。Scott Hanselman展示了使用它進行 API 呼叫是多么容易
您可以使用以下方法直接檢索資料GetBytesAsync：
record CopyRequest(Uri sourceUri,Uri blobUri);
...
var requests=new List<CopyRequest>();
//Load some source/target URLs
var client=new HttpClient();
await Parallel.ForEachAsync(requests,async req=>{
var bytes=await client.GetBytesAsync(req.sourceUri);
var blob=new CloudAppendBlob(req.targetUri);
await blob.UploadFromByteArrayAsync(bytes, 0, bytes.Length);
});
更好的選擇是將資料作為流檢索并將其直接發送到 blob ：
await Parallel.ForEachAsync(requests,async req=>{
var response=await client.GetAsync(req.sourceUri,
HttpCompletionOption.ResponseHeadersRead);
using var sourceStream=await response.Content.ReadAsStreamAsync();
var blob=new CloudAppendBlob(req.targetUri);
await blob.UploadFromStreamAsync(sourceStream);
});
HttpCompletionOption.ResponseHeadersRead導致GetAsync收到回應標頭后立即回傳，而不緩沖任何回應資料。
.NET 3.1
In older .NET Core versions (which are reaching End-of-Life in a few months) you can use eg an ActionBlock with a Degree-Of-Parallelism greater than 1:
var options=new ExecuteDataflowBlockOptions{ MaxDegreeOfParallelism = 8};
var copyBlock=new ActionBlock<CopyRequest>(async req=>{
var response=await client.GetAsync(req.sourceUri,
HttpCompletionOption.ResponseHeadersRead);
using var sourceStream=await response.Content.ReadAsStreamAsync();
var blob=new CloudAppendBlob(req.targetUri);
await blob.UploadFromStreamAsync(sourceStream);
}, options);
The block classes in the TPL Dataflow library can be used to construct processing pipelines similar to a shell script pipeline, with each block piping its output to the next block.
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/455887.html
標籤：C＃ 。网 .net-core 任务并行库 并行.foreach