多线程foreach Api Post和write响应到一个文件夹

Question

我有一个api，我需要在其中发布请求并编写响应到单个txt文件。

每个post请求都需要一个我从列表中读取的MaterialID。这个列表可以包含1000到5000个materialid

我需要一种方法来做多个并行请求。

下面的代码是我目前拥有的，但是是为同步请求构建的。

如何处理并行请求?

//reads MaterialID from file
System.IO.StreamReader fileTXT =
new System.IO.StreamReader(@"C:\TempTest.txt");
while ((line = fileTXT.ReadLine()) != null)
{
Material_Count.Add(line.Trim());
UpdateLogTxt("Material_Count = " + line.ToString() + "  ");
}
fileTXT.Close();

//loops through MaterialID list and starts post request 
foreach (var item in Material_Count)
{
try
{
UpdateLogTxt(" Submitting   = " + item.ToString());

var Set = Properties.Settings.Default;
using (var httpClient = new HttpClient())
{
string FCSEndPoint = @"https://www.api.co.za";
string FCSToken = "";
FCSToken = textBoxDescription.Text.Trim();

string uri = $"{FCSEndPoint}/material/pageIndex=" + item.ToString() + "";
HttpResponseMessage response = null;
httpClient.BaseAddress = new Uri(uri);
httpClient.DefaultRequestHeaders.Accept.Clear();
httpClient.DefaultRequestHeaders.Accept.Add(new MediaTypeWithQualityHeaderValue("application/json"));
httpClient.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("SessionToken", FCSToken);
AuthenticationHeaderValue headerObj = new AuthenticationHeaderValue("SessionToken", FCSToken);
httpClient.DefaultRequestHeaders.Authorization = headerObj;
var TaskObj = httpClient.PostAsJsonAsync(uri, String.Empty);
TaskObj.Wait();
HttpResponseMessage messageOut = TaskObj.Result;
response = TaskObj.Result;

if (response.IsSuccessStatusCode)
{
var TaskObj2 = messageOut.Content.ReadAsStringAsync();
TaskObj2.Wait();
string ResponseStr = TaskObj2.Result;

//writes response to a txt file. 
string fileName = @"C:TempmaterialMaterial_ " + item.ToString() + " " + DateTime.Now.ToString("yyyyMMddmmss") + ".txt";

try
{
// Check if file already exists. If yes, delete it.     
if (File.Exists(fileName))
{
File.Delete(fileName);
}

// Create a new file     
using (StreamWriter sw = File.CreateText(fileName))
{
sw.WriteLine(ResponseStr.ToString());
}


}
catch (Exception ex)
{
UpdateLogTxt("Exception Write to file  Failed =  ---  " + ex.ToString());
}

}
else if (response.StatusCode == System.Net.HttpStatusCode.Forbidden)
{ 
UpdateLogTxt("Response Failed (Forbidden) =  ---  " + response.StatusCode.ToString());

}
else
{
}

}
}
catch (HttpRequestException h)
{
UpdateLogTxt("HttpRequestException Send Failed = ---  " + h.ToString());
}
catch (Exception ex)
{
UpdateLogTxt("Exception Send Failed = ---  " + ex.ToString());
}

}

Answer 1

将System.Threading.Tasks命名空间中的foreach替换为Parallel.ForEach

你可以在这里找到方便的示例代码=>平行的。ForEach方法

Answer 2

这里有一个使用微软的响应式框架- NuGetSystem.Reactive的替代方案，然后你可以这样做:

首先收集代码中使用的几个常量，以确保它们只在UI线程中发生一次:

string fcsToken = textBoxDescription.Text.Trim();
string now = DateTime.Now.ToString("yyyyMMddmmss");

现在让我们创建一个助手方法来处理HTTP调用:

Task<HttpResponseMessage> PostAsJsonAsync(HttpClient hc, string item)
{
string fcsEndPoint = @"https://www.api.co.za";
string uri = $"{fcsEndPoint}/material/pageIndex={item}";
hc.BaseAddress = new Uri(uri);
hc.DefaultRequestHeaders.Accept.Clear();
hc.DefaultRequestHeaders.Accept.Add(new MediaTypeWithQualityHeaderValue("application/json"));
hc.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("SessionToken", fcsToken);
AuthenticationHeaderValue authorization = new AuthenticationHeaderValue("SessionToken", fcsToken);
hc.DefaultRequestHeaders.Authorization = authorization;
return hc.PostAsJsonAsync(uri, String.Empty);
}

现在我们可以编写可观察对象查询:

IObservable<(string fileName, string responseText)> query =
from item in
File
.ReadLines(@"C:\TempTest.txt")
.Select(x => x.Trim())
.ToObservable()
from output in
Observable
.Using(
() => new HttpClient(),
hc =>
from response in Observable.FromAsync(() => PostAsJsonAsync(hc, item))
where response.IsSuccessStatusCode
from responseText in Observable.FromAsync(() => response.Content.ReadAsStringAsync())
select
(
fileName: $@"C:TempmaterialMaterial_ {item} {now}.txt",
responseText: responseText
))
select output;

最后是订阅:

IDisposable subscription =
query
.Subscribe(x => File.WriteAllText(x.fileName, x.responseText));

就是这样。都是异步和并发的