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

充凝

前言

在处理大型数据集时，单线程处理往往成为性能瓶颈。通过将数据分割成多个小块，并利用多线程进行并行处理，可以显著提升程序的执行效率和响应速度。
本文将详细介绍几种高效的多线程并行处理实现方式，帮助开发者优化数据处理流程。

使用Parallel.ForEach进行并行处理

最简单的实现方式是使用C#内置的

1. Parallel.ForEach

复制代码

方法。

1. namespace AppParallel
   
2. {
   
3.     internal class Program
   
4.     {
   
5.         static object lockObject =
   
6.         new object();
   
7.         static void Main(string[] args)
   
8.         {
   
9.             // 创建示例数据
   
10.             var largeList =
    
11.             Enumerable.Range(1, 1000000).ToList();
    
12. 
    
13.             // 设置并行选项
    
14.             var parallelOptions = new ParallelOptions
    
15.             {
    
16.                 MaxDegreeOfParallelism =
    
17.                 Environment.ProcessorCount
    
18.                 // 使用处理器核心数量的线程
    
19.             };
    
20. 
    
21.             try
    
22.             {
    
23.                 Parallel.ForEach(largeList, parallelOptions,
    
24.                 (number) =>
    
25.                 {
    
26.                     // 这里是对每个元素的处理逻辑
    
27.                     var result = ComplexCalculation(number);
    
28. 
    
29.                     // 注意：如果需要收集结果，要考虑线程安全
    
30.                     lock (lockObject)
    
31.                     {
    
32.                         // 进行线程安全的结果收集
    
33.                         Console.WriteLine(result);
    
34.                     }
    
35.                 });
    
36.             }
    
37.             catch (AggregateException ae)
    
38.             {
    
39.                 // 处理并行处理中的异常
    
40.                 foreach (var ex in
    
41.                 ae.InnerExceptions)
    
42.                 {
    
43.                     Console.WriteLine($"Error:
    
44.                     {ex.Message}");
    
45.                 }
    
46.             }
    
47.         }
    
48.         private static int
    
49.         ComplexCalculation(int number)
    
50.         {
    
51.             // 模拟复杂计算
    
52.             Thread.Sleep(100);
    
53.             return number * 2;
    
54.         }
    
55. 
    
56.     }
    
57. }

复制代码

手动分块处理方式

有时我们需要更精细的控制，可以手动将数据分块并分配给不同的线程。

1. namespace AppParallel
   
2. {
   
3.     internal class Program
   
4.     {
   
5.         static void Main(string[] args)
   
6.         {
   
7.             var largeList = Enumerable.Range(1, 1000000).ToList();
   
8.             ProcessByChunks(largeList, 1000);
   
9.             // 每1000个元素一个块
   
10.         }
    
11.         public static void ProcessByChunks<T>(List<T> largeList,
    
12.         int chunkSize)
    
13.         {
    
14.             // 计算需要多少个分块
    
15.             int chunksCount = (int)Math.Ceiling((double)largeList.Count / chunkSize);
    
16.             var tasks = new List<Task>();
    
17. 
    
18.             for (int i = 0; i < chunksCount; i++)
    
19.             {
    
20.                 // 获取当前分块的数据
    
21.                 var chunk = largeList
    
22.                     .Skip(i * chunkSize)
    
23.                     .Take(chunkSize)
    
24.                     .ToList();
    
25. 
    
26.                 // 创建新任务处理当前分块
    
27.                 var task = Task.Run(() => ProcessChunk(chunk));
    
28.                 tasks.Add(task);
    
29.             }
    
30. 
    
31.             // 等待所有任务完成
    
32.             Task.WaitAll(tasks.ToArray());
    
33.         }
    
34. 
    
35.         private static void
    
36.         ProcessChunk<T>(List<T> chunk)
    
37.         {
    
38.             foreach (var item in chunk)
    
39.             {
    
40.                 // 处理每个元素
    
41.                 ProcessItem(item);
    
42.             }
    
43.         }
    
44. 
    
45.         private static void
    
46.         ProcessItem<T>(T item)
    
47.         {
    
48.             // 具体的处理逻辑
    
49.             Console.WriteLine
    
50.             ($"Processing item: {item} on thread: {Task.CurrentId}");
    
51.         }
    
52. 
    
53.     }
    
54. }

复制代码

使用生产者-消费者模式

对于更复杂的场景，我们可以使用生产者-消费者模式，这样可以更好地控制内存使用和处理流程。

1. public class ProducerConsumerExample
   
2. {
   
3.     private readonly BlockingCollection<int> _queue;
   
4.     private readonly
   
5.     int _producerCount;
   
6.     private readonly
   
7.     int _consumerCount;
   
8.     private readonly
   
9.     CancellationTokenSource _cts;
   
10. 
    
11.     public ProducerConsumerExample(int queueCapacity = 1000)
    
12.     {
    
13.         _queue = new BlockingCollection<int>(queueCapacity);
    
14.         _producerCount = 1;
    
15.         _consumerCount =
    
16.         Environment.ProcessorCount;
    
17.         _cts = new CancellationTokenSource();
    
18.     }
    
19. 
    
20.     public async Task ProcessDataAsync(List<int> largeList)
    
21.     {
    
22.         // 创建生产者任务
    
23.         var producerTask =
    
24.         Task.Run(() => Producer(largeList));
    
25. 
    
26.         // 创建消费者任务
    
27.         var consumerTasks = Enumerable.Range(0, _consumerCount)
    
28.             .Select(_ => Task.Run(() => Consumer()))
    
29.             .ToList();
    
30. 
    
31.         // 等待所有生产者完成
    
32.         await producerTask;
    
33. 
    
34.         // 标记队列已完成
    
35.         _queue.CompleteAdding();
    
36. 
    
37.         // 等待所有消费者完成
    
38.         await Task.WhenAll(consumerTasks);
    
39.     }
    
40. 
    
41.     private void Producer(List<int> items)
    
42.     {
    
43.         try
    
44.         {
    
45.             foreach (var item in items)
    
46.             {
    
47.                 if (_cts.
    
48.                 Token.IsCancellationRequested)
    
49.                     break;
    
50. 
    
51.                 _queue.Add(item);
    
52.             }
    
53.         }
    
54.         catch (Exception ex)
    
55.         {
    
56.             Console.WriteLine($"Producer error:
    
57.             {ex.Message}");
    
58.             _cts.Cancel();
    
59.         }
    
60.     }
    
61. 
    
62.     private void Consumer()
    
63.     {
    
64.         try
    
65.         {
    
66.             foreach (var item in _queue.GetConsumingEnumerable())
    
67.             {
    
68.                 if (_cts.Token.IsCancellationRequested)
    
69.                     break;
    
70. 
    
71.                 // 处理数据
    
72.                 ProcessItem(item);
    
73.             }
    
74.         }
    
75.         catch (Exception ex)
    
76.         {
    
77.             Console.WriteLine($"Consumer error: {ex.Message}");
    
78.             _cts.Cancel();
    
79.         }
    
80.     }
    
81. 
    
82.     private void ProcessItem(int item)
    
83.     {
    
84.         // 具体的处理逻辑
    
85.         Thread.Sleep(100);
    
86.         // 模拟耗时操作
    
87.         Console.WriteLine($"Processed item {item} on thread {Task.CurrentId}");
    
88.     }
    
89. }
    
90. 
    
91. // 使用示例
    
92. static async Task Main(string[] args)
    
93. {
    
94.     var processor = new ProducerConsumerExample();
    
95.     var largeList = Enumerable.Range(1, 10000).ToList();
    
96.     await processor.ProcessDataAsync(largeList);
    
97. }

复制代码

注意事项

1、合适的分块大小：分块不宜过小，因为过多的线程切换会抵消并行处理的优势；也不宜过大，以免影响负载均衡。建议从每块1000到5000个元素开始测试，找到最优的分块大小。
2、异常处理：务必妥善处理并行处理中的异常情况。每个任务应使用try-catch语句包装，确保异常不会导致整个程序崩溃。同时，考虑使用CancellationToken来优雅地终止所有任务。
3、资源管理：注意内存使用，避免一次性加载过多数据。合理控制并发线程的数量，通常不超过处理器核心数的两倍。对于实现了IDisposable接口的资源，使用using语句进行管理，确保资源及时释放。
4、线程安全：访问共享资源时必须保证线程安全，可以使用适当的同步机制如锁（lock）、信号量（Semaphore）等。考虑使用线程安全的集合类，例如ConcurrentDictionary或ConcurrentQueue。避免过度锁定，以免造成性能瓶颈。
通过遵循这些注意事项，可以确保在C#中高效且安全地进行大数据列表的并行处理。

总结

并行处理大数据列表是提升程序性能的有效手段，但需根据具体场景选择合适的实现方式。
本文介绍了三种主要方法，各有其适用场景和优势：
Parallel.ForEach：适用于简单场景，易于实现且代码简洁。适合快速原型开发或处理逻辑较为直接的任务。
手动分块处理：提供更精细的控制，适合中等复杂度场景。允许开发者优化分块大小和线程分配，以达到最佳性能。
生产者-消费者模式：适用于复杂场景，能够更好地管理资源使用和任务调度。特别适合需要高效处理大量数据流或涉及多个处理阶段的应用。
在实际应用中，建议首先进行性能测试，根据数据量大小、处理复杂度以及系统的硬件配置选择最合适的实现方式。
另外，务必重视异常处理和资源管理，确保程序的稳定性和可靠性。通过合理的并行处理策略，可以显著提高大型数据集的处理效率，为应用程序带来更好的用户体验。
以上就是C#中高效的多线程并行处理实现方式详解的详细内容，更多关于C#多线程并行处理的资料请关注晓枫资讯其它相关文章！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！