2 回答

TA貢獻1806條經驗 獲得超8個贊
更新
正如我剛剛注意到您在評論中提到的那樣,問題是由數學計算引起的。
將計算和更新數據庫的部分分開會更好。
對于計算部分,使用Parallel.ForEach()以便優化您的工作,您可以控制線程數。
只有在所有這些任務完成之后。用于async-await將您的數據更新到數據庫,無需SemaphoreSlim我提及。
public static async Task<int> Work()
{
var id = await CreateIdInDB() // async create record in DB
// run background task, don't wait when it finishes
Task.Run(async () => {
//Calculation Part
ConcurrentBag<int> data = new ConcurrentBag<int>();
Parallel.ForEach(
listOfData,
new ParallelOptions { CancellationToken = token, MaxDegreeOfParallelism = 3 },
x => {ConcurrentBag.Add(calculationPart(x))});
//Update DB part
int[] data_arr = data.ToArray();
List<Task> worker = new List<Task>();
foreach (var i in data_arr)
{
worker.Add(DBPart(x));
}
await Task.WhenAll(worker);
});
// return created id immediately
return id;
}
當你async-await在Parallel.forEach.
首先,閱讀第一個和第二個答案的這個問題。將這兩者結合起來毫無意義。
實際上async-await會最大限度地利用可用線程,所以簡單地使用它。
public static async Task<int> Work()
{
var id = await CreateIdInDB() // async create record in DB
// run background task, don't wait when it finishes
Task.Run(async () => {
List<Task> worker = new List<Task>();
foreach (var i in listOfData)
{
worker.Add(ProcessSingle(x));
}
await Task.WhenAll(worker);
});
// return created id immediately
return id;
}
但是還有另一個問題,在這種情況下,這些任務仍然一起開始,消耗你的 CPU 使用率。
因此,為避免這種情況,請使用 SemaphoreSlim
public static async Task<int> Work()
{
var id = await CreateIdInDB() // async create record in DB
// run background task, don't wait when it finishes
Task.Run(async () => {
List<Task> worker = new List<Task>();
//To limit the number of Task started.
var throttler = new SemaphoreSlim(initialCount: 20);
foreach (var i in listOfData)
{
await throttler.WaitAsync();
worker.Add(Task.Run(async () =>
{
await ProcessSingle(x);
throttler.Release();
}
));
}
await Task.WhenAll(worker);
});
// return created id immediately
return id;
}
此外,不要Task.Factory.StartNew()在簡單Task.Run()就足以完成您想要的工作時使用,請閱讀 Stephen Cleary 撰寫的這篇出色的文章。

TA貢獻1830條經驗 獲得超3個贊
如果您更熟悉“傳統”并行處理概念,請像這樣重寫您的 ProcessSingle() 方法:
public static void ProcessSingle(MyInputData inputData)
{
var dbData = GetDataFromDb(); // get data from DB async using Dapper
// some lasting processing (sync)
SaveDataToDb(); // async save processed data to DB using Dapper
}
當然,您最好也以類似的方式更改 Work() 方法。
- 2 回答
- 0 關注
- 1152 瀏覽
添加回答
舉報