代碼思路
將要處理的數據放到ConcurrentQueue
中,然后開啟多個線程去處理數據,處理完成后,再到隊列中獲取下一個待處理數據。
ConcurrentQueue
表示線程安全的先進先出 (FIFO) 集合,屬於System.Collections.Concurrent
命名空間下的一個數據結構
直接上代碼
/// <summary>
/// 多線程處理數據(無返回值)
/// </summary>
/// <typeparam name="T">數據類型</typeparam>
/// <param name="list">待處理數據</param>
/// <param name="action">數據處理方法(有參數無返回值)</param>
/// <param name="count">處理線程數量</param>
/// <param name="waitFlag">是否等待執行結束</param>
static void RunTask<T>(List<T> list, Action<T> action, int threadCount = 5, bool waitFlag = true)
{
ConcurrentQueue<T> queue = new ConcurrentQueue<T>(list);
Task[] tasks = new Task[threadCount];
for (int i = 0; i < threadCount; i++)
{
tasks[i] = Task.Run(() =>
{
while (queue.TryDequeue(out T t))
{
action(t);
}
});
}
if (waitFlag)
{
Task.WaitAll(tasks);
}
}
/// <summary>
/// 多線程處理數據(返回處理后列表)
/// </summary>
/// <typeparam name="T">數據類型</typeparam>
/// <param name="list">待處理數據</param>
/// <param name="func">數據處理方法(有參數有返回值)</param>
/// <param name="threadCount">處理線程數量</param>
/// <returns>數據處理后結果</returns>
static List<T> RunTask<T>(List<T> list, Func<T, T> func, int threadCount = 5)
{
var result = new List<T>();
ConcurrentQueue<T> queue = new ConcurrentQueue<T>(list);
Task<List<T>>[] tasks = new Task<List<T>>[threadCount];
for (int i = 0; i < threadCount; i++)
{
tasks[i] = Task.Run<List<T>>(() =>
{
var rList = new List<T>();
while (queue.TryDequeue(out T t))
{
rList.Add(func(t));
}
return rList;
});
}
Task.WaitAll(tasks);
for (int i = 0; i < threadCount; i++)
{
result.AddRange(tasks[i].Result);
}
return result;
}
調用方法
List<int> list = new List<int>() { 1, 2, 3, 4, 5, 6, 7, 8, 9 };
//輸出列表中的數據,且加上“action”字符前綴
RunTask<int>(list, d => { Console.WriteLine("action" + d); });
//對列表中數據都執行 “*2” 的操作
var result = RunTask<int>(list, d => { return d * 2; });
result.ForEach(d => Console.WriteLine(d));
最后的話
上面的代碼只是簡單的實現了對數據的處理,並沒有考慮到對內存的使用限制,一般的項目中使用還是可以的。
最后我嘗試了下,生成一個100M的列表,然后將其加載到ConcurrentQueue
,監控程序內存占用,發現沒有很大的內存占用變化。