多線程編程系列之線程間通信和協(xié)作
一、 線程間通信的方式和實(shí)現(xiàn)
在多線程編程中，線程間通信是非常常見的需求，它指的是多個(gè)線程之間通過某種機(jī)制來交換信息，協(xié)調(diào)彼此的行為。線程間通信的方式常用的有以下幾種：
共享內(nèi)存：線程之間可以通過共享內(nèi)存來交換信息，每個(gè)線程擁有對(duì)共享內(nèi)存區(qū)域的讀寫權(quán)限。一般情況下需要使用鎖來保證共享內(nèi)存的同步與互斥。
using System.Threading;
class Program {
private static int data = 0;
private static bool isRunning = true;
static void Main(string[] args) {
using (var mutex = new Mutex(false, "sharedMutex"))
using (var mappedFile = MemoryMappedFile.CreateOrOpen("sharedMemory", 1024))
using (var accessor = mappedFile.CreateViewAccessor()) {
var thread1 = new Thread(() => {
while (isRunning) {
mutex.WaitOne();
accessor.Write(0, data);
mutex.ReleaseMutex();
Thread.Sleep(1000);
}
});
var thread2 = new Thread(() => {
while (isRunning) {
mutex.WaitOne();
data = accessor.ReadInt32(0);
mutex.ReleaseMutex();
Console.WriteLine($"Data received: {data}");
Thread.Sleep(1000);
}
});
thread1.Start();
thread2.Start();
Console.WriteLine("Press enter to exit.");
Console.ReadLine();
isRunning = false;
thread1.Join();
thread2.Join();
}
}
}該程序創(chuàng)建了一個(gè)名為 sharedMemory 的內(nèi)存映射文件和一個(gè)名為 sharedMutex 的互斥對(duì)象。兩個(gè)線程分別負(fù)責(zé)將數(shù)據(jù)寫入內(nèi)存映射文件和讀取內(nèi)存映射文件中的數(shù)據(jù)。在操作前，需要通過互斥對(duì)象進(jìn)行同步與互斥。
消息傳遞：線程之間可以通過發(fā)送消息來交換信息，每個(gè)線程擁有一個(gè)接收緩沖區(qū)和一個(gè)發(fā)送緩沖區(qū)。這里的消息指的是數(shù)據(jù)包或者數(shù)據(jù)流，線程之間通過操作緩沖區(qū)來完成數(shù)據(jù)交換。
using System.Collections.Concurrent;
class Program {
private static bool isRunning = true;
private static BlockingCollection<string> messageQueue = new BlockingCollection<string>();
static void Main(string[] args) {
var thread1 = new Thread(() => {
while (isRunning) {
var message = messageQueue.Take();
Console.WriteLine($"Thread 1 received message: {message}");
}
});
var thread2 = new Thread(() => {
while (isRunning) {
var message = messageQueue.Take();
Console.WriteLine($"Thread 2 received message: {message}");
}
});
thread1.Start();
thread2.Start();
// Simulate sending some messages
messageQueue.Add("Hello from thread 1");
messageQueue.Add("Hello from thread 2");
messageQueue.Add("Hello again from thread 1");
messageQueue.Add("Goodbye from thread 2");
Console.WriteLine("Press enter to exit.");
Console.ReadLine();
isRunning = false;
thread1.Join();
thread2.Join();
}
}該程序使用 BlockingCollection<T> 類來實(shí)現(xiàn)簡(jiǎn)單的消息隊(duì)列，兩個(gè)線程分別負(fù)責(zé)從消息隊(duì)列中取出消息并進(jìn)行處理。在發(fā)送消息時(shí)，可以將其添加到消息隊(duì)列中。
管道：線程之間可以通過管道來交換信息，一個(gè)線程將數(shù)據(jù)寫入管道，另一個(gè)線程則從管道中讀取數(shù)據(jù)。管道本質(zhì)上也是一種共享內(nèi)存的方式，并且會(huì)自動(dòng)進(jìn)行同步(管道的大小是有限制的)
信號(hào)量：線程之間可以通過信號(hào)量來同步和互斥訪問資源，一個(gè)線程獲取信號(hào)量后就可以進(jìn)行訪問操作，其他線程則需要等待。信號(hào)量可以用于實(shí)現(xiàn)進(jìn)程之間的同步和互斥，但在多線程應(yīng)用中使用時(shí)需要注意信號(hào)量的實(shí)現(xiàn)。
using System.Threading;
class Program {
private static bool isRunning = true;
private static AutoResetEvent signal = new AutoResetEvent(false);
static void Main(string[] args) {
var thread1 = new Thread(() => {
while (isRunning) {
Console.WriteLine("Thread 1 is waiting...");
signal.WaitOne();
Console.WriteLine("Thread 1 received signal.");
}
});
var thread2 = new Thread(() => {
while (isRunning) {
Console.WriteLine("Thread 2 is waiting...");
signal.WaitOne();
Console.WriteLine("Thread 2 received signal.");
}
});
thread1.Start();
thread2.Start();
// Send signals to the threads
signal.Set(); // signals only one of the waiting threads
signal.Set(); // signals the other waiting thread
Console.WriteLine("Press enter to exit.");
Console.ReadLine();
isRunning = false;
signal.Set(); // unblock any waiting threads
thread1.Join();
thread2.Join();
}
}該程序使用 AutoResetEvent 類來實(shí)現(xiàn)線程間的同步。兩個(gè)線程等待信號(hào)并進(jìn)行處理，主線程發(fā)送信號(hào)來通知等待的線程進(jìn)行處理?？梢允褂?Set() 方法發(fā)送信號(hào)并使用 WaitOne() 方法等待信號(hào)。
using System.Threading;
class Program {
private static bool isRunning = true;
private static ManualResetEvent signal = new ManualResetEvent(false);
static void Main(string[] args) {
var thread1 = new Thread(() => {
while (isRunning) {
Console.WriteLine("Thread 1 is waiting...");
signal.WaitOne();
Console.WriteLine("Thread 1 received signal.");
signal.Reset(); // reset the signal
}
});
var thread2 = new Thread(() => {
while (isRunning) {
Console.WriteLine("Thread 2 is waiting...");
signal.WaitOne();
Console.WriteLine("Thread 2 received signal.");
signal.Reset(); // reset the signal
}
});
thread1.Start();
thread2.Start();
// Send signals to the threads
signal.Set(); // signals both of the waiting threads
Console.WriteLine("Press enter to exit.");
Console.ReadLine();
isRunning = false;
signal.Set(); // unblock any waiting threads
thread1.Join();
thread2.Join();
}
}該程序使用 ManualResetEvent 類來實(shí)現(xiàn)線程間的同步。兩個(gè)線程等待信號(hào)并進(jìn)行處理，主線程發(fā)送信號(hào)來通知等待的線程進(jìn)行處理。可以使用 Set() 方法發(fā)送信號(hào)并使用 WaitOne() 方法等待信號(hào)，同時(shí)使用 Reset() 方法將信號(hào)狀態(tài)重置為未發(fā)出狀態(tài)，以便下次等待。
互斥鎖：線程之間可以通過互斥鎖來同步和互斥訪問共享資源，一個(gè)線程獲取鎖后就可以進(jìn)行訪問操作，其他線程則需要等待釋放鎖?；コ怄i是一種經(jīng)典的同步和互斥機(jī)制，在多線程編程中用得比較廣泛，一般和條件變量一起使用。
using System.Threading;
class Program {
private static bool isRunning = true;
private static object lockObject = new object();
private static int counter = 0;
static void Main(string[] args) {
var thread1 = new Thread(() => {
while (isRunning) {
lock (lockObject) {
while (counter % 2 == 1) {
Monitor.Wait(lockObject);
}
Console.WriteLine($"Thread 1: {counter++}");
Monitor.PulseAll(lockObject);
}
}
});
var thread2 = new Thread(() => {
while (isRunning) {
lock (lockObject) {
while (counter % 2 == 0) {
Monitor.Wait(lockObject);
}
Console.WriteLine($"Thread 2: {counter++}");
Monitor.PulseAll(lockObject);
}
}
});
thread1.Start();
thread2.Start();
Console.WriteLine("Press enter to exit.");
Console.ReadLine();
isRunning = false;
thread1.Join();
thread2.Join();
}
}二、同步和異步線程間通信的比較
同步和異步線程間通信的主要區(qū)別在于調(diào)用者是否需要等待被調(diào)用者完成任務(wù)才能繼續(xù)執(zhí)行下一步操作。
同步線程間通信指的是調(diào)用者主動(dòng)向被調(diào)用者請(qǐng)求一個(gè)任務(wù)，并等待被調(diào)用者完成后再繼續(xù)執(zhí)行。這種模式對(duì)于簡(jiǎn)單的應(yīng)用程序來說很容易實(shí)現(xiàn)，但有時(shí)會(huì)引發(fā)線程死鎖的問題，因?yàn)槿绻鄠€(gè)線程都在等待對(duì)方完成任務(wù)，就會(huì)形成死循環(huán)。
異步線程間通信則是被調(diào)用者在處理任務(wù)的同時(shí)，通知調(diào)用者任務(wù)的狀態(tài)。這種模式可以提高程序的響應(yīng)速度，因?yàn)檎{(diào)用者可以繼續(xù)執(zhí)行其他任務(wù)，而不必等待被調(diào)用者完成任務(wù)才能進(jìn)行下一步操作。
C# 語言提供了多種方式來實(shí)現(xiàn)線程間的同步和異步通信。其中，同步通信可以使用 Mutex、Semaphore 和 Monitor 等互斥量類來實(shí)現(xiàn)線程鎖定和等待，在獲取到資源后再釋放鎖定。異步通信可以使用委托、事件和 Completion 是C# 5.0 開始的異步編程功能，可以使用 async 和 await 關(guān)鍵字來快速實(shí)現(xiàn)異步編程。
三、 多個(gè)線程協(xié)作完成任務(wù)
在多線程編程中，有時(shí)我們需要多個(gè)線程協(xié)作完成一個(gè)復(fù)雜的任務(wù)。這些線程需要互相通信、協(xié)調(diào)以達(dá)到同一目標(biāo)。下面是一些常用的多線程協(xié)作技術(shù)：
信號(hào)量 Semaphore：Semaphore 可以用來控制某一資源的訪問權(quán)，比如網(wǎng)絡(luò)連接數(shù)限制、數(shù)據(jù)庫連接池等。Semaphore 通過計(jì)數(shù)器來控制資源的數(shù)量，并提供了 Acquire 和 Release 等方法來允許或阻塞線程訪問資源。多個(gè)線程可以共享一個(gè) Semaphore，當(dāng) Semaphore 計(jì)數(shù)為 0 時(shí)，其他線程就需要等待。
Mutex：Mutex 是一種操作系統(tǒng)提供的同步機(jī)制，它可以保證在同一時(shí)刻只有一個(gè)線程訪問共享資源。Mutex 提供了 Lock 和 Unlock 等方法來保護(hù)臨界區(qū)。如果一個(gè)線程獲得 Mutex，其他線程就必須等待直到該線程釋放 Mutex。
AutoResetEvent 和 ManualResetEvent：這兩種事件用于線程間的同步，AutoResetEvent 的 WaitOne 方法會(huì)阻塞當(dāng)前線程直到事件被發(fā)出，發(fā)出后事件重置為未發(fā)出狀態(tài)；ManualResetEvent 則不會(huì)自動(dòng)重置，需要調(diào)用 Reset 方法手動(dòng)將事件重置為未發(fā)出狀態(tài)。
CountdownEvent：CountdownEvent通常用于多個(gè)線程都需要完成某個(gè)任務(wù)后才能繼續(xù)執(zhí)行的場(chǎng)景。當(dāng)所有線程都完成任務(wù)后，調(diào)用Done方法通知CountdownEvent，等待的線程就會(huì)被喚醒。
