C#中高效的多線程并行處理實現方式詳解

更新時間：2025年04月07日 10:12:03 作者：技術老小子

在處理大型數據集時,單線程處理往往成為性能瓶頸,所以本文將詳細介紹幾種高效的多線程并行處理實現方式,幫助開發(fā)者優(yōu)化數據處理流程,有需要的可以了解下

前言

在處理大型數據集時，單線程處理往往成為性能瓶頸。通過將數據分割成多個小塊，并利用多線程進行并行處理，可以顯著提升程序的執(zhí)行效率和響應速度。

本文將詳細介紹幾種高效的多線程并行處理實現方式，幫助開發(fā)者優(yōu)化數據處理流程。

使用Parallel.ForEach進行并行處理

最簡單的實現方式是使用C#內置的Parallel.ForEach方法。

namespace AppParallel
{
    internal class Program
    {
        static object lockObject = 
        new object();
        static void Main(string[] args)
        {
            // 創(chuàng)建示例數據
            var largeList = 
            Enumerable.Range(1, 1000000).ToList();

            // 設置并行選項
            var parallelOptions = new ParallelOptions
            {
                MaxDegreeOfParallelism = 
                Environment.ProcessorCount 
                // 使用處理器核心數量的線程
            };

            try
            {
                Parallel.ForEach(largeList, parallelOptions,
                (number) =>
                {
                    // 這里是對每個元素的處理邏輯
                    var result = ComplexCalculation(number);

                    // 注意：如果需要收集結果，要考慮線程安全
                    lock (lockObject)
                    {
                        // 進行線程安全的結果收集
                        Console.WriteLine(result);
                    }
                });
            }
            catch (AggregateException ae)
            {
                // 處理并行處理中的異常
                foreach (var ex in 
                ae.InnerExceptions)
                {
                    Console.WriteLine($"Error: 
                    {ex.Message}");
                }
            }
        }
        private static int
        ComplexCalculation(int number)
        {
            // 模擬復雜計算
            Thread.Sleep(100);
            return number * 2;
        }

    }
}

手動分塊處理方式

有時我們需要更精細的控制，可以手動將數據分塊并分配給不同的線程。

namespace AppParallel
{
    internal class Program
    {
        static void Main(string[] args)
        {
            var largeList = Enumerable.Range(1, 1000000).ToList();
            ProcessByChunks(largeList, 1000); 
            // 每1000個元素一個塊
        }
        public static void ProcessByChunks<T>(List<T> largeList, 
        int chunkSize)
        {
            // 計算需要多少個分塊
            int chunksCount = (int)Math.Ceiling((double)largeList.Count / chunkSize);
            var tasks = new List<Task>();

            for (int i = 0; i < chunksCount; i++)
            {
                // 獲取當前分塊的數據
                var chunk = largeList
                    .Skip(i * chunkSize)
                    .Take(chunkSize)
                    .ToList();

                // 創(chuàng)建新任務處理當前分塊
                var task = Task.Run(() => ProcessChunk(chunk));
                tasks.Add(task);
            }

            // 等待所有任務完成
            Task.WaitAll(tasks.ToArray());
        }

        private static void 
        ProcessChunk<T>(List<T> chunk)
        {
            foreach (var item in chunk)
            {
                // 處理每個元素
                ProcessItem(item);
            }
        }

        private static void 
        ProcessItem<T>(T item)
        {
            // 具體的處理邏輯
            Console.WriteLine
            ($"Processing item: {item} on thread: {Task.CurrentId}");
        }

    }
}

使用生產者-消費者模式

對于更復雜的場景，我們可以使用生產者-消費者模式，這樣可以更好地控制內存使用和處理流程。

public class ProducerConsumerExample
{
    private readonly BlockingCollection<int> _queue;
    private readonly 
    int _producerCount;
    private readonly 
    int _consumerCount;
    private readonly 
    CancellationTokenSource _cts;

    public ProducerConsumerExample(int queueCapacity = 1000)
    {
        _queue = new BlockingCollection<int>(queueCapacity);
        _producerCount = 1;
        _consumerCount = 
        Environment.ProcessorCount;
        _cts = new CancellationTokenSource();
    }

    public async Task ProcessDataAsync(List<int> largeList)
    {
        // 創(chuàng)建生產者任務
        var producerTask = 
        Task.Run(() => Producer(largeList));

        // 創(chuàng)建消費者任務
        var consumerTasks = Enumerable.Range(0, _consumerCount)
            .Select(_ => Task.Run(() => Consumer()))
            .ToList();

        // 等待所有生產者完成
        await producerTask;

        // 標記隊列已完成
        _queue.CompleteAdding();

        // 等待所有消費者完成
        await Task.WhenAll(consumerTasks);
    }

    private void Producer(List<int> items)
    {
        try
        {
            foreach (var item in items)
            {
                if (_cts.
                Token.IsCancellationRequested)
                    break;

                _queue.Add(item);
            }
        }
        catch (Exception ex)
        {
            Console.WriteLine($"Producer error: 
            {ex.Message}");
            _cts.Cancel();
        }
    }

    private void Consumer()
    {
        try
        {
            foreach (var item in _queue.GetConsumingEnumerable())
            {
                if (_cts.Token.IsCancellationRequested)
                    break;

                // 處理數據
                ProcessItem(item);
            }
        }
        catch (Exception ex)
        {
            Console.WriteLine($"Consumer error: {ex.Message}");
            _cts.Cancel();
        }
    }

    private void ProcessItem(int item)
    {
        // 具體的處理邏輯
        Thread.Sleep(100);
        // 模擬耗時操作
        Console.WriteLine($"Processed item {item} on thread {Task.CurrentId}");
    }
}

// 使用示例
static async Task Main(string[] args)
{
    var processor = new ProducerConsumerExample();
    var largeList = Enumerable.Range(1, 10000).ToList();
    await processor.ProcessDataAsync(largeList);
}