C#中高效的多线程并行处理实现方式详解_C#

前言

在处理大型数据集时，单线程处理往往成为性能瓶颈。通过将数据分割成多个小块，并利用多线程进行并行处理，可以显著提升程序的执行效率和响应速度。

本文将详细介绍几种高效的多线程并行处理实现方式，帮助开发者优化数据处理流程。

使用parallel.foreach进行并行处理

最简单的实现方式是使用c#内置的parallel.foreach方法。

namespace appparallel
{
    internal class program
    {
        static object lockobject = 
        new object();
        static void main(string[] args)
        {
            // 创建示例数据
            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();

                // 创建新任务处理当前分块
                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)
    {
        // 创建生产者任务
        var producertask = 
        task.run(() => producer(largelist));

        // 创建消费者任务
        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);
}