登录/ 注册
 找回密码
 立即注册
关闭

CSDN热搜

程序园 精品问答 技术交流 资源下载
返回列表 发新帖
首页 › 资源区 › 代码 ›C#编程中并行与并发的简单理解

C#编程中并行与并发的简单理解

吕梓美 2025-5-29 10:53:11
1.简述

并发通过管理多个任务的执行顺序，确保系统在高负载下仍能保持响应性；并行则利用多处理器或多核心硬件，真正同时执行任务，以加速计算。这两者在高性能计算、实时系统和用户交互应用中发挥着不可替代的作用。
在多核处理器时代，传统串行编程已无法充分利用硬件潜力。并行计算通过将任务分解到多个核心执行，显著缩短了计算时间。然而，并发与并行的实现并非没有代价，它们引入了诸如竞争条件、死锁和负载均衡等复杂问题，需要开发者具备深厚的理论基础和实践经验。
2.并发与并行

2.1 定义


  • 并发（Concurrency）：

    • 指系统在一段时间内管理多个任务的能力。并发关注任务的协调与交错执行，通过时间分片等技术在一个或多个处理器上实现，因此并发看似同时进行，但不一定在同一时刻执行。
    • 并发强调任务的逻辑组织和协调。
    • 举例：一个Web服务器可以并发处理多个客户端请求，通过快速切换任务确保每个请求都能及时响应。

  • 并行（Parallelism）：

    • 指多个任务在同一时刻真正同时执行，通常依赖于多核处理器或分布式系统。其核心目标是提升计算速度，通过将问题分解为独立的子任务并同时处理。并行适用于计算密集型任务。
    • 并行关注物理执行的并行性。
    • 举例：在并行矩阵乘法中，不同的核心可以同时计算矩阵的不同部分，从而显著缩短总计算时间；科学模拟或图像处理，其效果依赖于多核处理器、GPU或分布式计算系统的硬件支持。

2.2 区别

1.png

并发与并行的根本区别在于执行的时间性和资源依赖性：

  • 执行模式：并行强调真正的同时执行，而并发通过任务切换营造同时进行的假象。
  • 硬件依赖：并行需要多处理器或多核心支持，而并发在单核系统上即可实现。
  • 目标：并行旨在加速计算，而并发注重系统响应性和多任务处理能力。
例如，在单核系统中，操作系统通过时间片轮转调度多个线程；而多核系统中，线程可以分配到不同核心并行运行。
3.实现并发

3.1 并行实现并发

在多核处理器上，任务可以分配到不同核心并行执行，从而实现高效并发。例如，Web服务器通过多线程并行处理客户端请求。
代码示例：多线程并行处理 ：System.Threading.ThreadPool来创建和管理线程池，并使用ManualResetEventSlim来等待所有任务完成。
  1.         using System;
  2.         using System.Collections.Generic;
  3.         using System.Threading;
  5.         class Program
  6.         {
  7.                 static void Main(string[] args)
  8.                 {
  9.                         List<Request> requests = new List<Request>
  10.                         {
  11.                                 new Request { Data = "Request1" },
  12.                                 new Request { Data = "Request2" },
  13.                                 new Request { Data = "Request3" }
  14.                                 // 添加更多请求
  15.                         };
  17.                         process_requests(requests);
  19.                         Console.WriteLine("All requests processed.");
  20.                 }
  22.                 static void process_requests(List<Request> requests)
  23.                 {
  24.                         int num_cores = Environment.ProcessorCount;  // 获取处理器核心数
  25.                         ManualResetEventSlim[] mres = new ManualResetEventSlim[requests.Count];  // 创建信号量数组
  27.                         for (int i = 0; i < requests.Count; i++)
  28.                         {
  29.                                 int index = i;
  30.                                 mres[index] = new ManualResetEventSlim(false);  // 初始化信号量
  31.                                 ThreadPool.QueueUserWorkItem((state) =>
  32.                                 {
  33.                                         handle_request(requests[index]);
  34.                                         mres[index].Set();  // 任务完成时设置信号量
  35.                                 });
  36.                         }
  38.                         // 等待所有任务完成
  39.                         ManualResetEventSlim.WaitAll(mres);
  40.                 }
  42.                 static void handle_request(Request request)
  43.                 {
  44.                         Response response = process(request);  // 处理请求
  45.                         send_response(response);  // 发送响应
  46.                 }
  48.                 static Response process(Request request)
  49.                 {
  50.                         // 模拟请求处理逻辑
  51.                         Console.WriteLine($"Processing request: {request.Data}");
  52.                         Thread.Sleep(1000);  // 模拟耗时操作
  53.                         return new Response { Data = $"Response for {request.Data}" };
  54.                 }
  56.                 static void send_response(Response response)
  57.                 {
  58.                         // 模拟发送响应逻辑
  59.                         Console.WriteLine($"Sending response: {response.Data}");
  60.                 }
  61.         }
  63.         class Request
  64.         {
  65.                 public string Data { get; set; }
  66.         }
  68.         class Response
  69.         {
  70.                 public string Data { get; set; }
  71.         }
复制代码
======================================================================================================================
使用Task.Run和Task.WhenAll来实现
  1.         using System;
  2.         using System.Collections.Generic;
  3.         using System.Threading.Tasks;
  5.         class Program
  6.         {
  7.                 static async Task Main(string[] args)
  8.                 {
  9.                         List<Request> requests = new List<Request>
  10.                         {
  11.                                 new Request { Data = "Request1" },
  12.                                 new Request { Data = "Request2" },
  13.                                 new Request { Data = "Request3" }
  14.                                 // 添加更多请求
  15.                         };
  17.                         await process_requests(requests);
  19.                         Console.WriteLine("All requests processed.");
  20.                 }
  22.                 static async Task process_requests(List<Request> requests)
  23.                 {
  24.                         List<Task> tasks = new List<Task>();
  26.                         foreach (Request request in requests)
  27.                         {
  28.                                 Task task = Task.Run(() => handle_request(request));
  29.                                 tasks.Add(task);
  30.                         }
  32.                         // 等待所有任务完成
  33.                         await Task.WhenAll(tasks);
  34.                 }
  36.                 static async Task handle_request(Request request)
  37.                 {
  38.                         Response response = await process(request);  // 处理请求
  39.                         send_response(response);  // 发送响应
  40.                 }
  42.                 static async Task<Response> process(Request request)
  43.                 {
  44.                         // 模拟请求处理逻辑
  45.                         Console.WriteLine($"Processing request: {request.Data}");
  46.                         await Task.Delay(1000);  // 模拟耗时操作
  47.                         return new Response { Data = $"Response for {request.Data}" };
  48.                 }
  50.                 static void send_response(Response response)
  51.                 {
  52.                         // 模拟发送响应逻辑
  53.                         Console.WriteLine($"Sending response: {response.Data}");
  54.                 }
  55.         }
  57.         class Request
  58.         {
  59.                 public string Data { get; set; }
  60.         }
  62.         class Response
  63.         {
  64.                 public string Data { get; set; }
  65.         }
复制代码
3.2 任务调度

在单核处理器上，通过时间片轮转等调度算法实现并发。操作系统在任务间快速切换，营造同时执行的假象。
代码示例：时间片轮转调度 ：示例使用了Task和CancellationToken来管理任务的时间片轮转调度。
  1.         using System;
  2.         using System.Collections.Generic;
  3.         using System.Threading;
  4.         using System.Threading.Tasks;
  6.         class Program
  7.         {
  8.                 static async Task Main(string[] args)
  9.                 {
  10.                         List<Task> tasks = new List<Task>
  11.                         {
  12.                                 run_task("Task1", 5000),  // 创建一个任务，模拟总时间为5秒
  13.                                 run_task("Task2", 3000),  // 创建一个任务，模拟总时间为3秒
  14.                                 run_task("Task3", 7000)   // 创建一个任务，模拟总时间为7秒
  15.                         };
  17.                         int time_slice = 1000;  // 设置时间片为1秒
  18.                         await scheduler(tasks, time_slice);
  20.                         Console.WriteLine("All tasks processed.");
  21.                 }
  23.                 static async Task scheduler(List<Task> tasks, int time_slice)
  24.                 {
  25.                         List<Task> runningTasks = new List<Task>();
  26.                         List<Task> remainingTasks = new List<Task>(tasks);
  28.                         while (remainingTasks.Count > 0 || runningTasks.Count > 0)
  29.                         {
  30.                                 // 将剩余任务中的第一个任务移到运行列表
  31.                                 if (remainingTasks.Count > 0)
  32.                                 {
  33.                                         runningTasks.Add(remainingTasks[0]);
  34.                                         remainingTasks.RemoveAt(0);
  35.                                 }
  37.                                 // 复制运行任务列表以避免在遍历过程中修改列表
  38.                                 List<Task> currentRunningTasks = new List<Task>(runningTasks);
  40.                                 foreach (Task task in currentRunningTasks)
  41.                                 {
  42.                                         if (!task.IsCompleted)
  43.                                         {
  44.                                                 await run_task_for_time_slice(task, time_slice);
  46.                                                 if (task.IsCompleted)
  47.                                                 {
  48.                                                         runningTasks.Remove(task);
  49.                                                 }
  50.                                                 else
  51.                                                 {
  52.                                                         remainingTasks.Add(task);
  53.                                                         runningTasks.Remove(task);
  54.                                                 }
  55.                                         }
  56.                                 }
  57.                         }
  58.                 }
  60.                 static async Task run_task_for_time_slice(Task task, int time_slice)
  61.                 {
  62.                         // 创建一个取消令牌源
  63.                         CancellationTokenSource cts = new CancellationTokenSource(time_slice);
  64.                         try
  65.                         {
  66.                                 // 等待任务完成或时间片用完
  67.                                 await task.WaitAsync(cts.Token);
  68.                         }
  69.                         catch (TaskCanceledException)
  70.                         {
  71.                                 // 时间片用完，任务未完成
  72.                                 Console.WriteLine($"Task {task.Id} preempted after {time_slice} ms");
  73.                         }
  74.                 }
  76.                 static Task run_task(string taskName, int total_time)
  77.                 {
  78.                         return Task.Run(async () =>
  79.                         {
  80.                                 int elapsedTime = 0;
  81.                                 int time_slice = 1000;  // 模拟内部时间片
  83.                                 while (elapsedTime < total_time)
  84.                                 {
  85.                                         Console.WriteLine($"{taskName} is running. Elapsed time: {elapsedTime} ms");
  86.                                         await Task.Delay(time_slice);  // 模拟任务运行一段时间
  87.                                         elapsedTime += time_slice;
  88.                                 }
  90.                                 Console.WriteLine($"{taskName} is completed.");
  91.                         });
  92.                 }
  93.         }
复制代码
3.3 多线程

多线程通过创建多个执行单元实现并发。线程共享进程资源，通过同步机制（如互斥锁）协调访问。
代码示例：多线程同步
  1.         using System;
  2.         using System.Collections.Generic;
  3.         using System.Threading;
  4.         using System.Threading.Tasks;
  6.         namespace Test.EventBus
  7.         {
  8.                 public class DemoB
  9.                 {
  10.                         private static Mutex mutex = new Mutex();  // 创建互斥锁
  11.                         private static StringBuilder sharedData = new StringBuilder();  // 初始化共享数据
  13.                         public void ShowMsg(string name, string msg)
  14.                         {
  15.                                 Console.WriteLine($"ShowMsg=> name:{name},msg:{msg}");
  16.                                 var eventMsg = new EventMessage
  17.                                 {
  18.                                         Name = name,
  19.                                         Msg = msg,
  20.                                         CreatedDate = DateTime.Now
  21.                                 };
  22.                                 EventPublishSubscribeUtils.PublishEvent(eventMsg, nameof(DemoB.ShowMsg));
  23.                         }
  25.                         public static void RunDemo()
  26.                         {
  27.                                 List<Task> tasks = new List<Task>();
  29.                                 for (int i = 0; i < 10; i++)
  30.                                 {
  31.                                         int taskId = i;
  32.                                         tasks.Add(Task.Run(() => thread_function($"Task{taskId}")));
  33.                                 }
  35.                                 // 等待所有任务完成
  36.                                 Task.WaitAll(tasks);
  37.                         }
  39.                         static void thread_function(string name)
  40.                         {
  41.                                 for (int i = 0; i < 5; i++)
  42.                                 {
  43.                                         modify_shared_data(name, i);
  44.                                 }
  45.                         }
  47.                         static void modify_shared_data(string name, int iteration)
  48.                         {
  49.                                 mutex.WaitOne();  // 加锁
  50.                                 try
  51.                                 {
  52.                                         // 修改共享数据
  53.                                         sharedData.AppendLine($"{name} is modifying shared data at iteration {iteration} on {DateTime.Now}");
  54.                                         Console.WriteLine($"{name} is modifying shared data at iteration {iteration} on {DateTime.Now}");
  55.                                 }
  56.                                 finally
  57.                                 {
  58.                                         mutex.ReleaseMutex();  // 解锁
  59.                                 }
  60.                         }
  61.                 }
  63.                 public class EventMessage
  64.                 {
  65.                         public string Name { get; set; }
  66.                         public string Msg { get; set; }
  67.                         public DateTime CreatedDate { get; set; }
  68.                 }
  70.                 public static class EventPublishSubscribeUtils
  71.                 {
  72.                         public static void PublishEvent(EventMessage eventMsg, string eventName)
  73.                         {
  74.                                 Console.WriteLine($"Published event: {eventName} => Name: {eventMsg.Name}, Msg: {eventMsg.Msg}, CreatedDate: {eventMsg.CreatedDate}");
  75.                         }
  76.                 }
  78.                 class Program
  79.                 {
  80.                         static void Main(string[] args)
  81.                         {
  82.                                 DemoB.RunDemo();
  83.                                 Console.WriteLine("All threads completed.");
  84.                         }
  85.                 }
  86.         }
复制代码
3.4 异步编程

异步编程通过事件循环和回调函数处理I/O密集型任务，避免阻塞主线程。
代码示例：异步I/O
  1. using System;
  2. using System.IO;
  3. using System.Threading.Tasks;
  5. namespace AsyncIOExample
  6. {
  7.         class Program
  8.         {
  9.                 static async Task Main(string[] args)
  10.                 {
  11.                         string filePath1 = "example1.txt";
  12.                         string filePath2 = "example2.txt";
  14.                         // 创建模拟文件
  15.                         File.WriteAllText(filePath1, "Data from example1.txt");
  16.                         File.WriteAllText(filePath2, "Data from example2.txt");
  18.                         // 异步读取文件并使用回调函数处理数据
  19.                         await async_read(filePath1, data => callback(data, filePath1));
  20.                         await async_read(filePath2, data => callback(data, filePath2));
  22.                         Console.WriteLine("All asynchronous read operations completed.");
  23.                 }
  25.                 static async Task async_read(string file, Action<string> callback)
  26.                 {
  27.                         // 模拟事件循环添加任务
  28.                         Console.WriteLine($"Starting asynchronous read for file: {file}");
  29.                         string data = await read_file(file);
  30.                         callback(data);
  31.                 }
  33.                 static async Task<string> read_file(string file)
  34.                 {
  35.                         // 模拟从磁盘读取文件
  36.                         using (StreamReader reader = new StreamReader(file))
  37.                         {
  38.                                 string data = await reader.ReadToEndAsync();
  39.                                 return data;
  40.                         }
  41.                 }
  43.                 static void callback(string data, string file)
  44.                 {
  45.                         // 处理读取后的数据
  46.                         Console.WriteLine($"Data read from {file}: {data}");
  47.                 }
  48.         }
  49. }
复制代码
3.5 协程

协程通过yield和resume机制在单线程内实现并发，适用于I/O密集型任务，具有低开销优势。
代码示例：协程
  1. using System;
  2. using System.Collections.Generic;
  3. using System.Threading.Tasks;
  5. namespace CoroutineExample
  6. {
  7.         class Program
  8.         {
  9.                 static async Task Main(string[] args)
  10.                 {
  11.                         IAsyncEnumerable<string> coroutine = coroutine_example();
  13.                         // 创建一个异步枚举器
  14.                         IAsyncEnumerator<string> enumerator = coroutine.GetAsyncEnumerator();
  16.                         // 启动协程
  17.                         if (await enumerator.MoveNextAsync())
  18.                         {
  19.                                 Console.WriteLine("Coroutine started.");
  21.                                 // 发送数据并恢复执行
  22.                                 await enumerator.MoveNextAsync();
  23.                                 enumerator.Current = "Data1";
  25.                                 await enumerator.MoveNextAsync();
  26.                                 enumerator.Current = "Data2";
  28.                                 await enumerator.MoveNextAsync();
  29.                                 enumerator.Current = "Data3";
  31.                                 // 结束协程
  32.                                 await enumerator.DisposeAsync();
  33.                         }
  34.                 }
  36.                 static async IAsyncEnumerable<string> coroutine_example()
  37.                 {
  38.                         string data = null;
  39.                         while (true)
  40.                         {
  41.                                 // 暂停并接收数据
  42.                                 await Task.Delay(100);  // 模拟等待
  43.                                 data = yield return data;
  45.                                 // 处理数据
  46.                                 process(data);
  47.                         }
  48.                 }
  50.                 static void process(string data)
  51.                 {
  52.                         if (data != null)
  53.                         {
  54.                                 Console.WriteLine($"Processed data: {data}");
  55.                         }
  56.                         else
  57.                         {
  58.                                 Console.WriteLine("No data to process.");
  59.                         }
  60.                 }
  61.         }
  62. }
复制代码
3.6 事件驱动

事件驱动编程通过事件循环监听和处理事件，适用于GUI和网络应用。
代码示例：事件驱动
  1. using System;
  2. using System.Collections.Generic;
  3. using System.Threading.Tasks;
  5. namespace EventDrivenExample
  6. {
  7.         class Program
  8.         {
  9.                 static async Task Main(string[] args)
  10.                 {
  11.                         // 创建事件循环
  12.                         EventLoop eventLoop = new EventLoop();
  14.                         // 注册事件处理函数
  15.                         eventLoop.RegisterHandler("Event1", Event1Handler);
  16.                         eventLoop.RegisterHandler("Event2", Event2Handler);
  18.                         // 模拟事件触发
  19.                         eventLoop.TriggerEvent(new Event { Type = "Event1", Data = "Data for Event1" });
  20.                         eventLoop.TriggerEvent(new Event { Type = "Event2", Data = "Data for Event2" });
  22.                         // 启动事件循环
  23.                         await eventLoop.Start();
  25.                         Console.WriteLine("Event loop completed.");
  26.                 }
  28.                 static void Event1Handler(Event e)
  29.                 {
  30.                         Console.WriteLine($"Handling {e.Type} with data: {e.Data}");
  31.                 }
  33.                 static void Event2Handler(Event e)
  34.                 {
  35.                         Console.WriteLine($"Handling {e.Type} with data: {e.Data}");
  36.                 }
  37.         }
  39.         public class Event
  40.         {
  41.                 public string Type { get; set; }
  42.                 public string Data { get; set; }
  43.         }
  45.         public class EventLoop
  46.         {
  47.                 private Queue<Event> _eventQueue = new Queue<Event>();
  48.                 private Dictionary<string, Action<Event>> _handlers = new Dictionary<string, Action<Event>>();
  49.                 private bool _running = false;
  51.                 public void RegisterHandler(string eventType, Action<Event> handler)
  52.                 {
  53.                         if (_handlers.ContainsKey(eventType))
  54.                         {
  55.                                 _handlers[eventType] += handler;
  56.                         }
  57.                         else
  58.                         {
  59.                                 _handlers[eventType] = handler;
  60.                         }
  61.                 }
  63.                 public void TriggerEvent(Event e)
  64.                 {
  65.                         lock (_eventQueue)
  66.                         {
  67.                                 _eventQueue.Enqueue(e);
  68.                         }
  69.                 }
  71.                 public async Task Start()
  72.                 {
  73.                         _running = true;
  74.                         while (_running)
  75.                         {
  76.                                 Event e = null;
  77.                                 lock (_eventQueue)
  78.                                 {
  79.                                         if (_eventQueue.Count > 0)
  80.                                         {
  81.                                                 e = _eventQueue.Dequeue();
  82.                                         }
  83.                                 }
  85.                                 if (e != null)
  86.                                 {
  87.                                         if (_handlers.TryGetValue(e.Type, out Action<Event> handler))
  88.                                         {
  89.                                                 handler(e);
  90.                                         }
  91.                                         else
  92.                                         {
  93.                                                 Console.WriteLine($"No handler registered for event type: {e.Type}");
  94.                                         }
  95.                                 }
  96.                                 else
  97.                                 {
  98.                                         // 模拟等待事件
  99.                                         await Task.Delay(100);  // 等待100毫秒
  100.                                 }
  101.                         }
  102.                 }
  104.                 public void Stop()
  105.                 {
  106.                         _running = false;
  107.                 }
  108.         }
  109. }
复制代码
3.7 多进程

多进程通过创建独立进程实现并发，进程间通过IPC（如管道或消息队列）通信，适用于CPU密集型任务
在C#中，多进程可以通过使用 System.Diagnostics.Process 类来创建和管理独立进程。进程间通信（IPC）可以通过多种方式实现，例如使用命名管道（System.IO.Pipes）或内存映射文件（System.IO.MemoryMappedFiles）。在这个示例中，我们将使用命名管道来进行进程间通信。
代码示例：多进程
  1. using System;
  2. using System.Collections.Generic;
  3. using System.Diagnostics;
  4. using System.IO.Pipes;
  5. using System.Text;
  6. using System.Threading.Tasks;
  8. namespace MultiProcessExample
  9. {
  10.         class Program
  11.         {
  12.                 static async Task Main(string[] args)
  13.                 {
  14.                         int num_processes = 3;  // 设置进程数量
  15.                         List<Process> processes = new List<Process>();
  16.                         List<Task<string>> readTasks = new List<Task<string>>();
  18.                         // 创建命名管道服务器
  19.                         var server = new NamedPipeServerStream("testpipe", PipeDirection.In, num_processes, PipeTransmissionMode.Message, PipeOptions.Asynchronous);
  21.                         // 创建并启动进程
  22.                         for (int i = 0; i < num_processes; i++)
  23.                         {
  24.                                 Process p = new Process();
  25.                                 p.StartInfo.FileName = "dotnet";
  26.                                 p.StartInfo.Arguments = $"MultiProcessExample.dll worker {i}";
  27.                                 p.StartInfo.UseShellExecute = false;
  28.                                 p.StartInfo.RedirectStandardOutput = true;
  29.                                 p.StartInfo.CreateNoWindow = true;
  30.                                 p.Start();
  32.                                 processes.Add(p);
  34.                                 // 读取子进程的输出
  35.                                 readTasks.Add(Task.Run(() => read_from_process(p)));
  36.                         }
  38.                         // 等待所有进程结束
  39.                         foreach (var process in processes)
  40.                         {
  41.                                 process.WaitForExit();
  42.                         }
  44.                         // 等待所有读取任务完成
  45.                         string[] results = await Task.WhenAll(readTasks);
  47.                         // 输出所有结果
  48.                         foreach (var result in results)
  49.                         {
  50.                                 Console.WriteLine($"Received result: {result}");
  51.                         }
  53.                         // 关闭命名管道服务器
  54.                         server.Close();
  55.                 }
  57.                 static string read_from_process(Process process)
  58.                 {
  59.                         // 读取子进程的标准输出
  60.                         string result = process.StandardOutput.ReadToEnd();
  61.                         return result;
  62.                 }
  63.         }
  65.         class Worker
  66.         {
  67.                 static async Task Main(string[] args)
  68.                 {
  69.                         if (args.Length != 2 || args[0] != "worker" || !int.TryParse(args[1], out int id))
  70.                         {
  71.                                 Console.WriteLine("Invalid arguments.");
  72.                                 return;
  73.                         }
  75.                         // 创建命名管道客户端
  76.                         using (NamedPipeClientStream pipeClient = new NamedPipeClientStream(".", "testpipe", PipeDirection.Out, PipeOptions.Asynchronous))
  77.                         {
  78.                                 try
  79.                                 {
  80.                                         // 连接到命名管道服务器
  81.                                         await pipeClient.ConnectAsync();
  83.                                         // 执行计算任务
  84.                                         string result = compute(id);
  86.                                         // 发送结果
  87.                                         send_result(pipeClient, result);
  88.                                 }
  89.                                 catch (Exception ex)
  90.                                 {
  91.                                         Console.WriteLine($"Error: {ex.Message}");
  92.                                 }
  93.                         }
  94.                 }
  96.                 static string compute(int id)
  97.                 {
  98.                         // 模拟计算任务
  99.                         Console.WriteLine($"Worker {id} is computing...");
  100.                         Task.Delay(1000).Wait();  // 模拟耗时操作
  101.                         return $"Result from Worker {id}";
  102.                 }
  104.                 static void send_result(NamedPipeClientStream pipeClient, string result)
  105.                 {
  106.                         try
  107.                         {
  108.                                 // 将结果发送到命名管道
  109.                                 byte[] resultBytes = Encoding.UTF8.GetBytes(result);
  110.                                 pipeClient.Write(resultBytes, 0, resultBytes.Length);
  111.                                 pipeClient.Flush();
  112.                         }
  113.                         catch (Exception ex)
  114.                         {
  115.                                 Console.WriteLine($"Error sending result: {ex.Message}");
  116.                         }
  117.                 }
  118.         }
  119. }
复制代码
4.实现并行的技术

4.1 多线程（Multithreading）

多线程通过在单个或多个处理器核心上运行多个线程来实现并行。在多核处理器上，线程可以真正并行执行；在单核处理器上，通过时间片切换实现伪并行。多线程适用于I/O密集型和计算密集型任务，能提高资源利用率和程序响应速度。
代码示例：：使用了System.Threading.Thread来创建和管理多个线程，并使用Task来提交和等待任务的完成。
  1.         using System;
  2.         using System.Collections.Generic;
  3.         using System.Threading;
  4.         using System.Threading.Tasks;
  6.         namespace MultiThreadExample
  7.         {
  8.                 class Program
  9.                 {
  10.                         static void Main(string[] args)
  11.                         {
  12.                                 int N = 3;  // 设置线程数量
  13.                                 List<Thread> threads = new List<Thread>();
  14.                                 List<string> results = new List<string>();
  15.                                 object lockObject = new object();  // 同步锁
  17.                                 // 创建并启动多个线程
  18.                                 for (int i = 0; i < N; i++)
  19.                                 {
  20.                                         int id = i;
  21.                                         Thread thread = new Thread(() => task_function(id, results, lockObject));
  22.                                         threads.Add(thread);
  23.                                         thread.Start();
  24.                                 }
  26.                                 // 等待所有线程完成
  27.                                 foreach (Thread thread in threads)
  28.                                 {
  29.                                         thread.Join();
  30.                                 }
  32.                                 // 输出所有结果
  33.                                 foreach (string result in results)
  34.                                 {
  35.                                         Console.WriteLine($"Result from thread: {result}");
  36.                                 }
  38.                                 Console.WriteLine("All threads completed.");
  39.                         }
  41.                         static void task_function(int id, List<string> results, object lockObject)
  42.                         {
  43.                                 string result = perform_task(id);  // 执行任务
  44.                                 lock (lockObject)
  45.                                 {
  46.                                         results.Add(result);  // 将结果添加到共享列表并加锁
  47.                                 }
  48.                         }
  50.                         static string perform_task(int id)
  51.                         {
  52.                                 // 模拟任务执行
  53.                                 Console.WriteLine($"Thread {id} is processing.");
  54.                                 Thread.Sleep(1000);  // 模拟耗时操作
  55.                                 return $"Result from Thread {id}";
  56.                         }
  57.                 }
  58.         }
复制代码
使用 Task 和 async/await 实现：
  1.         using System;
  2.         using System.Collections.Generic;
  3.         using System.Threading.Tasks;
  5.         namespace MultiThreadExample
  6.         {
  7.                 class Program
  8.                 {
  9.                         static async Task Main(string[] args)
  10.                         {
  11.                                 int N = 3;  // 设置线程数量
  12.                                 List<Task<string>> tasks = new List<Task<string>>();
  14.                                 // 创建并启动多个线程
  15.                                 for (int i = 0; i < N; i++)
  16.                                 {
  17.                                         int id = i;
  18.                                         Task<string> task = Task.Run(() => task_function(id));
  19.                                         tasks.Add(task);
  20.                                 }
  22.                                 // 等待所有线程完成
  23.                                 string[] results = await Task.WhenAll(tasks);
  25.                                 // 输出所有结果
  26.                                 foreach (string result in results)
  27.                                 {
  28.                                         Console.WriteLine($"Result from task: {result}");
  29.                                 }
  31.                                 Console.WriteLine("All tasks completed.");
  32.                         }
  34.                         static string task_function(int id)
  35.                         {
  36.                                 string result = perform_task(id);  // 执行任务
  37.                                 return result;
  38.                         }
  40.                         static string perform_task(int id)
  41.                         {
  42.                                 // 模拟任务执行
  43.                                 Console.WriteLine($"Task {id} is processing.");
  44.                                 Task.Delay(1000).Wait();  // 模拟耗时操作
  45.                                 return $"Result from Task {id}";
  46.                         }
  47.                 }
  48.         }
复制代码
4.2 多进程（Multiprocessing）

多进程通过创建多个独立进程实现并行，每个进程运行在不同的处理器核心上。进程间通过管道或消息队列等通信机制协调工作。多进程适用于需要高隔离性和安全性的任务，如科学计算和服务器应用。
代码示例：：
  1.         using System;
  2.         using System.Collections.Generic;
  3.         using System.Diagnostics;
  4.         using System.IO.Pipes;
  5.         using System.Text;
  6.         using System.Threading.Tasks;
  8.         namespace MultiProcessExample
  9.         {
  10.                 class Program
  11.                 {
  12.                         static async Task Main(string[] args)
  13.                         {
  14.                                 int N = 3;  // 设置进程数量
  15.                                 List<Process> processes = new List<Process>();
  16.                                 List<Task<string>> readTasks = new List<Task<string>>();
  18.                                 // 创建命名管道服务器
  19.                                 using (NamedPipeServerStream pipeServer = new NamedPipeServerStream("testpipe", PipeDirection.In, N, PipeTransmissionMode.Message, PipeOptions.Asynchronous))
  20.                                 {
  21.                                         // 创建并启动多个进程
  22.                                         for (int i = 0; i < N; i++)
  23.                                         {
  24.                                                 Process process = create_process(i);
  25.                                                 processes.Add(process);
  26.                                                 process.Start();
  28.                                                 // 创建一个任务来读取子进程的结果
  29.                                                 readTasks.Add(Task.Run(() => read_from_pipe(pipeServer)));
  30.                                         }
  32.                                         // 等待所有进程完成
  33.                                         foreach (var process in processes)
  34.                                         {
  35.                                                 process.WaitForExit();
  36.                                         }
  38.                                         // 等待所有读取任务完成
  39.                                         string[] results = await Task.WhenAll(readTasks);
  41.                                         // 输出所有结果
  42.                                         foreach (var result in results)
  43.                                         {
  44.                                                 Console.WriteLine($"Received result: {result}");
  45.                                         }
  47.                                         // 关闭命名管道服务器
  48.                                         pipeServer.Close();
  49.                                 }
  51.                                 Console.WriteLine("All processes completed.");
  52.                         }
  54.                         static Process create_process(int id)
  55.                         {
  56.                                 Process process = new Process();
  57.                                 process.StartInfo.FileName = "dotnet";
  58.                                 process.StartInfo.Arguments = $"MultiProcessExample.dll worker {id}";
  59.                                 process.StartInfo.UseShellExecute = false;
  60.                                 process.StartInfo.RedirectStandardOutput = true;
  61.                                 process.StartInfo.CreateNoWindow = true;
  62.                                 return process;
  63.                         }
  65.                         static async Task<string> read_from_pipe(NamedPipeServerStream pipeServer)
  66.                         {
  67.                                 // 等待客户端连接
  68.                                 await pipeServer.WaitForConnectionAsync();
  70.                                 // 创建字节数组来接收数据
  71.                                 byte[] buffer = new byte[1024];
  72.                                 int bytesRead = await pipeServer.ReadAsync(buffer, 0, buffer.Length);
  73.                                 string result = Encoding.UTF8.GetString(buffer, 0, bytesRead);
  75.                                 // 断开连接
  76.                                 pipeServer.Disconnect();
  78.                                 return result;
  79.                         }
  80.                 }
  81.         }
复制代码
**使用Worker类：
**
  1.         using System;
  2.         using System.IO.Pipes;
  3.         using System.Threading.Tasks;
  5.         namespace MultiProcessExample
  6.         {
  7.                 class Worker
  8.                 {
  9.                         static async Task Main(string[] args)
  10.                         {
  11.                                 if (args.Length != 2 || args[0] != "worker" || !int.TryParse(args[1], out int id))
  12.                                 {
  13.                                         Console.WriteLine("Invalid arguments.");
  14.                                         return;
  15.                                 }
  17.                                 // 创建命名管道客户端
  18.                                 using (NamedPipeClientStream pipeClient = new NamedPipeClientStream(".", "testpipe", PipeDirection.Out, PipeOptions.Asynchronous))
  19.                                 {
  20.                                         try
  21.                                         {
  22.                                                 // 连接到命名管道服务器
  23.                                                 await pipeClient.ConnectAsync();
  25.                                                 // 执行计算任务
  26.                                                 string result = compute(id);
  28.                                                 // 发送结果
  29.                                                 send_result(pipeClient, result);
  30.                                         }
  31.                                         catch (Exception ex)
  32.                                         {
  33.                                                 Console.WriteLine($"Error: {ex.Message}");
  34.                                         }
  35.                                 }
  36.                         }
  38.                         static string compute(int id)
  39.                         {
  40.                                 // 模拟计算任务
  41.                                 Console.WriteLine($"Worker {id} is computing...");
  42.                                 Task.Delay(1000).Wait();  // 模拟耗时操作
  43.                                 return $"Result from Worker {id}";
  44.                         }
  46.                         static void send_result(NamedPipeClientStream pipeClient, string result)
  47.                         {
  48.                                 try
  49.                                 {
  50.                                         // 将结果发送到命名管道
  51.                                         byte[] resultBytes = Encoding.UTF8.GetBytes(result);
  52.                                         pipeClient.Write(resultBytes, 0, resultBytes.Length);
  53.                                         pipeClient.Flush();
  54.                                 }
  55.                                 catch (Exception ex)
  56.                                 {
  57.                                         Console.WriteLine($"Error sending result: {ex.Message}");
  58.                                 }
  59.                         }
  60.                 }
  61.         }
复制代码
4.3 分布式计算（Distributed Computing）

分布式计算将任务分配到网络中的多台计算机上并行执行，通常使用消息传递接口（MPI）进行通信。适用于大规模数据处理和复杂计算任务，如天气预报和分布式数据库。
为了简化实现，我们可以使用一个简单的消息传递库，例如 NamedPipes 和 Task 来模拟MPI的行为。这里我们使用 NamedPipes 来进行进程间通信，并模拟主节点和工作节点之间的数据交换。
代码示例：
  1.         using System;
  2.         using System.Collections.Generic;
  3.         using System.Diagnostics;
  4.         using System.IO.Pipes;
  5.         using System.Text;
  6.         using System.Threading.Tasks;
  8.         namespace DistributedComputingExample
  9.         {
  10.                 class Program
  11.                 {
  12.                         static async Task Main(string[] args)
  13.                         {
  14.                                 int num_workers = 3;  // 设置工作节点数量
  15.                                 List<Process> workers = new List<Process>();
  16.                                 List<Task<string>> readTasks = new List<Task<string>>();
  18.                                 // 创建和启动工作节点
  19.                                 for (int i = 1; i <= num_workers; i++)
  20.                                 {
  21.                                         Process worker = create_worker_process(i);
  22.                                         workers.Add(worker);
  23.                                         worker.Start();
  24.                                 }
  26.                                 // 模拟主节点
  27.                                 if (args.Length == 0 || args[0] != "worker")
  28.                                 {
  29.                                         // 主节点逻辑
  30.                                         string data = load_data(num_workers);
  31.                                         Console.WriteLine("Data loaded.");
  33.                                         // 创建命名管道服务器来发送数据
  34.                                         List<NamedPipeServerStream> sendPipes = new List<NamedPipeServerStream>();
  35.                                         for (int i = 1; i <= num_workers; i++)
  36.                                         {
  37.                                                 NamedPipeServerStream sendPipe = new NamedPipeServerStream($"sendpipe_{i}", PipeDirection.Out, 1, PipeTransmissionMode.Message, PipeOptions.Asynchronous);
  38.                                                 sendPipes.Add(sendPipe);
  39.                                         }
  41.                                         // 发送数据到每个工作节点
  42.                                         for (int i = 1; i <= num_workers; i++)
  43.                                         {
  44.                                                 string data_chunk = data.Split('|')[i - 1];
  45.                                                 send_data(sendPipes[i - 1], data_chunk);
  46.                                         }
  48.                                         // 创建命名管道服务器来接收结果
  49.                                         List<NamedPipeServerStream> receivePipes = new List<NamedPipeServerStream>();
  50.                                         for (int i = 1; i <= num_workers; i++)
  51.                                         {
  52.                                                 NamedPipeServerStream receivePipe = new NamedPipeServerStream($"receivepipe_{i}", PipeDirection.In, 1, PipeTransmissionMode.Message, PipeOptions.Asynchronous);
  53.                                                 receivePipes.Add(receivePipe);
  54.                                         }
  56.                                         // 读取每个工作节点的结果
  57.                                         for (int i = 1; i <= num_workers; i++)
  58.                                         {
  59.                                                 readTasks.Add(Task.Run(() => read_from_pipe(receivePipes[i - 1])));
  60.                                         }
  62.                                         // 等待所有工作节点完成
  63.                                         foreach (var worker in workers)
  64.                                         {
  65.                                                 worker.WaitForExit();
  66.                                         }
  68.                                         // 等待所有读取任务完成
  69.                                         string[] results = await Task.WhenAll(readTasks);
  71.                                         // 聚合结果
  72.                                         string final_result = aggregate(results);
  73.                                         Console.WriteLine($"Final result: {final_result}");
  74.                                 }
  75.                                 else
  76.                                 {
  77.                                         // 工作节点逻辑
  78.                                         int id = int.Parse(args[1]);
  79.                                         Console.WriteLine($"Worker {id} started.");
  81.                                         // 创建命名管道客户端来接收数据
  82.                                         using (NamedPipeClientStream receivePipe = new NamedPipeClientStream(".", $"sendpipe_{id}", PipeDirection.In, PipeOptions.Asynchronous))
  83.                                         {
  84.                                                 await receivePipe.ConnectAsync();
  85.                                                 string data_chunk = receive_data(receivePipe);
  86.                                                 Console.WriteLine($"Worker {id} received data: {data_chunk}");
  88.                                                 // 处理数据
  89.                                                 string result = process(data_chunk);
  90.                                                 Console.WriteLine($"Worker {id} processed data: {result}");
  92.                                                 // 创建命名管道客户端来发送结果
  93.                                                 using (NamedPipeClientStream sendPipe = new NamedPipeClientStream(".", $"receivepipe_{id}", PipeDirection.Out, PipeOptions.Asynchronous))
  94.                                                 {
  95.                                                         await sendPipe.ConnectAsync();
  96.                                                         send_result(sendPipe, result);
  97.                                                 }
  98.                                         }
  99.                                 }
  100.                         }
  102.                         static Process create_worker_process(int id)
  103.                         {
  104.                                 Process process = new Process();
  105.                                 process.StartInfo.FileName = "dotnet";
  106.                                 process.StartInfo.Arguments = $"DistributedComputingExample.dll worker {id}";
  107.                                 process.StartInfo.UseShellExecute = false;
  108.                                 process.StartInfo.RedirectStandardOutput = true;
  109.                                 process.StartInfo.CreateNoWindow = true;
  110.                                 return process;
  111.                         }
  113.                         static string load_data(int num_chunks)
  114.                         {
  115.                                 // 模拟加载数据
  116.                                 string data = "DataChunk1|DataChunk2|DataChunk3";
  117.                                 return data;
  118.                         }
  120.                         static void send_data(NamedPipeServerStream pipe, string data)
  121.                         {
  122.                                 try
  123.                                 {
  124.                                         byte[] dataBytes = Encoding.UTF8.GetBytes(data);
  125.                                         pipe.Write(dataBytes, 0, dataBytes.Length);
  126.                                         pipe.Flush();
  127.                                         pipe.Disconnect();
  128.                                 }
  129.                                 catch (Exception ex)
  130.                                 {
  131.                                         Console.WriteLine($"Error sending data: {ex.Message}");
  132.                                 }
  133.                         }
  135.                         static string receive_data(NamedPipeClientStream pipe)
  136.                         {
  137.                                 try
  138.                                 {
  139.                                         byte[] buffer = new byte[1024];
  140.                                         int bytesRead = pipe.Read(buffer, 0, buffer.Length);
  141.                                         return Encoding.UTF8.GetString(buffer, 0, bytesRead);
  142.                                 }
  143.                                 catch (Exception ex)
  144.                                 {
  145.                                         Console.WriteLine($"Error receiving data: {ex.Message}");
  146.                                         return null;
  147.                                 }
  148.                         }
  150.                         static string process(string data_chunk)
  151.                         {
  152.                                 // 模拟任务处理
  153.                                 Console.WriteLine($"Processing data chunk: {data_chunk}");
  154.                                 Task.Delay(1000).Wait();  // 模拟耗时操作
  155.                                 return $"Processed {data_chunk}";
  156.                         }
  158.                         static void send_result(NamedPipeClientStream pipe, string result)
  159.                         {
  160.                                 try
  161.                                 {
  162.                                         byte[] resultBytes = Encoding.UTF8.GetBytes(result);
  163.                                         pipe.Write(resultBytes, 0, resultBytes.Length);
  164.                                         pipe.Flush();
  165.                                 }
  166.                                 catch (Exception ex)
  167.                                 {
  168.                                         Console.WriteLine($"Error sending result: {ex.Message}");
  169.                                 }
  170.                         }
  172.                         static async Task<string> read_from_pipe(NamedPipeServerStream pipe)
  173.                         {
  174.                                 await pipe.WaitForConnectionAsync();
  175.                                 byte[] buffer = new byte[1024];
  176.                                 int bytesRead = await pipe.ReadAsync(buffer, 0, buffer.Length);
  177.                                 string result = Encoding.UTF8.GetString(buffer, 0, bytesRead);
  178.                                 pipe.Disconnect();
  179.                                 return result;
  180.                         }
  182.                         static string aggregate(string[] results)
  183.                         {
  184.                                 // 聚合结果
  185.                                 StringBuilder finalResult = new StringBuilder();
  186.                                 foreach (string result in results)
  187.                                 {
  188.                                         finalResult.AppendLine(result);
  189.                                 }
  190.                                 return finalResult.ToString();
  191.                         }
  192.                 }
  193.         }
复制代码
5 实践运用

5.1 软件开发中的并行应用

并行广泛应用于需要高计算能力的场景，包括：

  • 科学模拟：天气预报、分子动力学等任务涉及大量方程求解，可通过并行化显著加速。
  • 机器学习：深度神经网络训练依赖矩阵运算，TensorFlow和PyTorch等框架利用GPU并行性加速训练过程。
  • 图像与视频处理：如3D渲染或视频滤镜应用，可将任务分配到多核或GPU上并行执行。
常见的并行编程模型包括：

  • T- PL：TPL是.NET中用于并行编程的一个强大库
  • OpenMP：基于指令的共享内存并行API，适用于C/C++和Fortran。
  • MPI（消息传递接口）：分布式内存并行的标准，用于高性能计算集群。
  • CUDA：NVIDIA的并行计算平台，支持GPU上的细粒度并行。
5.2 软件开发中的并发应用

并发在需要处理多任务或事件的系统中至关重要，例如：

  • Web服务器：如Apache和Nginx，通过多线程、多进程或事件驱动架构并发处理大量客户端请求。
  • 图形用户界面（GUI）：并发确保界面在执行后台任务（如数据加载）时仍能响应用户输入。
  • 数据库系统：通过锁和事务等并发控制机制，管理多用户对数据的并发访问。
常见的并发模型包括：

  • 多线程：C#、Java和C++提供线程库（如System.Thread、java.lang.Thread、std::thread）实现并发。
  • 异步编程：Node.js和Python的asyncio支持非阻塞代码，适用于I/O密集型任务。
  • Actor模型：Erlang和Akka框架通过独立的Actor单元和消息传递实现并发，避免共享内存问题。
6. 并发与并行编程的挑战

6.1 并发挑战

并发引入了多个复杂问题：

  • 竞争条件（Race Conditions）：多个线程同时访问共享资源，可能导致不可预测的结果。例如，未同步的计数器递增可能丢失更新。
  • 死锁（Deadlocks）：线程间相互等待对方释放资源，导致永久阻塞。例如，两个线程各自持有对方需要的锁。
  • 活锁（Livelocks）：线程不断尝试解决问题但无进展，如反复让出资源。
  • 饥饿（Starvation）：某些线程因调度不公而无法获得资源。
解决这些问题通常依赖同步原语（如互斥锁、信号量），但过度同步可能降低性能。
6.2 并行挑战

并行计算也有其难点：

  • 负载均衡：确保所有处理器或核心均匀分担工作量，避免部分核心空闲。
  • 通信开销：分布式系统中，节点间通信成本可能抵消并行收益。
  • 可扩展性：随着处理器数量增加，同步开销或串行部分可能导致收益递减。
并行算法需精心设计，采用动态负载均衡或工作窃取等技术应对这些挑战。
7. 管理并行与并发的工具与技术

7.1 调试与测试

并发与并行程序的非确定性使其调试异常困难，常用工具包括：

  • 静态分析：如Intel Inspector或FindBugs，可在不运行代码的情况下检测潜在问题。
  • 运行时验证：Valgrind的Helgrind等工具在程序运行时监控同步错误。
  • 测试框架：JUnit或pytest可扩展用于并发测试，模拟多线程场景。
7.2 设计模式

设计模式为常见问题提供解决方案：

  • 线程池：管理固定数量的线程执行任务，减少创建和销毁开销。
  • 生产者-消费者：生产者生成数据，消费者处理数据，通过同步队列协调。
  • Map-Reduce：将任务映射到数据分片并归约结果，适用于大数据处理。
7.3 编程语言支持

现代语言内置了对并行与并发的支持：

  • CSharp：通过TPL和System.Collections.Concurrent等库简化并发和并行编程。
  • Go：通过goroutines和通道简化并发编程。
  • Rust：通过所有权模型在编译时防止数据竞争。
  • Java：提供java.util.concurrent包，包括线程池、并发集合等高级工具。
8.并行与并发的权衡

8.1 复杂度与性能

并行与并发提升性能的同时增加了代码复杂度：

  • 多线程：提供细粒度控制，但易引入竞争条件。
  • 异步编程：避免线程开销，但可能导致回调地狱或复杂逻辑。
8.2 共享内存与消息传递

并发模型分为两种：

  • 共享内存：线程共享数据，需同步以避免冲突，效率高但易出错。
  • 消息传递：通过消息通信避免共享状态，安全性高但可能引入延迟。
如何选择取决于性能、安全性和应用需求。

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

使用道具 举报
相关推荐
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册
发帖
吕梓美
2025-5-29 10:53:11

0

粉丝关注

11

主题发布

板块介绍填写区域，请于后台编辑
微信扫一扫