這篇文章主要介紹了一個(gè)進(jìn)程間通訊同步的C#框架,代碼具有相當(dāng)?shù)姆€(wěn)定性和可維護(hù)性,隨著.NET的開(kāi)源也會(huì)被注入更多活力,推薦!需要的朋友可以參考下

0.背景簡(jiǎn)介

微軟在 .NET 框架中提供了多種實(shí)用的線程同步手段，其中包括 monitor 類(lèi)及 reader-writer鎖。但跨進(jìn)程的同步方法還是非常欠缺。另外，目前也沒(méi)有方便的線程間及進(jìn)程間傳遞消息的方法。例如C/S和SOA，又或者生產(chǎn)者/消費(fèi)者模式中就常常需要傳遞消息。為此我編寫(xiě)了一個(gè)獨(dú)立完整的框架，實(shí)現(xiàn)了跨線程和跨進(jìn)程的同步和通訊。這框架內(nèi)包含了信號(hào)量，信箱，內(nèi)存映射文件，阻塞通道，及簡(jiǎn)單消息流控制器等組件。這篇文章里提到的類(lèi)同屬于一個(gè)開(kāi)源的庫(kù)項(xiàng)目(BSD許可)，你可以從這里下載到 www.cdrnet.net/projects/threadmsg/.

這個(gè)框架的目的是：

封裝性：通過(guò)MSMQ消息隊(duì)列發(fā)送消息的線程無(wú)需關(guān)心消息是發(fā)送到另一個(gè)線程還是另一臺(tái)機(jī)器。

簡(jiǎn)單性：向其他進(jìn)程發(fā)送消息只需調(diào)用一個(gè)方法。

注意：我刪除了本文中全部代碼的XML注釋以節(jié)省空間。如果你想知道這些方法和參數(shù)的詳細(xì)信息，請(qǐng)參考附件中的代碼。

1.先看一個(gè)簡(jiǎn)單例子

使用了這個(gè)庫(kù)后，跨進(jìn)程的消息傳遞將變得非常簡(jiǎn)單。我將用一個(gè)小例子來(lái)作示范：一個(gè)控制臺(tái)程序，根據(jù)參數(shù)可以作為發(fā)送方也可以作為接收方運(yùn)行。在發(fā)送程序里，你可以輸入一定的文本并發(fā)送到信箱內(nèi)(返回key)，接收程序?qū)@示所有從信箱內(nèi)收到的消息。你可以運(yùn)行無(wú)數(shù)個(gè)發(fā)送程序和接收程序，但是每個(gè)消息只會(huì)被具體的某一個(gè)接收程序所收到。

1. [Serializable] 
2. struct Message 
3. { 
4. public string Text; 
5. } 
6.  
7. class Test 
8. { 
9. IMailBox mail; 
10.  
11. public Test() 
12. { 
13. mail = new ProcessMailBox("TMProcessTest",1024); 
14. } 
15.  
16. public void RunWriter() 
17. { 
18. Console.WriteLine("Writer started"); 
19. Message msg; 
20. while(true) 
21. { 
22. msg.Text = Console.ReadLine(); 
23. if(msg.Text.Equals("exit")) 
24. break; 
25. mail.Content = msg; 
26. } 
27. } 
28.  
29. public void RunReader() 
30. { 
31. Console.WriteLine("Reader started"); 
32. while(true) 
33. { 
34. Message msg = (Message)mail.Content; 
35. Console.WriteLine(msg.Text); 
36. } 
37. } 
38.  
39. [STAThread] 
40. static void Main(string[] args) 
41. { 
42. Test test = new Test(); 
43. if(args.Length > 0) 
44. test.RunWriter(); 
45. else 
46. test.RunReader(); 
47. } 
48. } 

信箱一旦創(chuàng)建之后(這上面代碼里是 ProcessMailBox )，接收消息只需要讀取 Content 屬性，發(fā)送消息只需要給這個(gè)屬性賦值。當(dāng)沒(méi)有數(shù)據(jù)時(shí)，獲取消息將會(huì)阻塞當(dāng)前線程;發(fā)送消息時(shí)如果信箱里已經(jīng)有數(shù)據(jù)，則會(huì)阻塞當(dāng)前線程。正是有了這個(gè)阻塞，整個(gè)程序是完全基于中斷的，并且不會(huì)過(guò)度占用CPU(不需要進(jìn)行輪詢(xún))。發(fā)送和接收的消息可以是任意支持序列化(Serializable)的類(lèi)型。

然而，實(shí)際上暗地里發(fā)生的事情有點(diǎn)復(fù)雜：消息通過(guò)內(nèi)存映射文件來(lái)傳遞，這是目前唯一的跨進(jìn)程共享內(nèi)存的方法，這個(gè)例子里我們只會(huì)在 pagefile 里面產(chǎn)生虛擬文件。對(duì)這個(gè)虛擬文件的訪問(wèn)是通過(guò) win32 信號(hào)量來(lái)確保同步的。消息首先序列化成二進(jìn)制，然后再寫(xiě)進(jìn)該文件，這就是為什么需要聲明Serializable屬性。內(nèi)存映射文件和 win32 信號(hào)量都需要調(diào)用 NT內(nèi)核的方法。多得了 .NET 框架中的 Marshal 類(lèi)，我們可以避免編寫(xiě)不安全的代碼。我們將在下面討論更多的細(xì)節(jié)。

2. .NET里面的跨線程/進(jìn)程同步

線程/進(jìn)程間的通訊需要共享內(nèi)存或者其他內(nèi)建機(jī)制來(lái)發(fā)送/接收數(shù)據(jù)。即使是采用共享內(nèi)存的方式，也還需要一組同步方法來(lái)允許并發(fā)訪問(wèn)。

同一個(gè)進(jìn)程內(nèi)的所有線程都共享公共的邏輯地址空間(堆)。對(duì)于不同進(jìn)程，從 win2000 開(kāi)始就已經(jīng)無(wú)法共享內(nèi)存。然而，不同的進(jìn)程可以讀寫(xiě)同一個(gè)文件。WinAPI提供了多種系統(tǒng)調(diào)用方法來(lái)映射文件到進(jìn)程的邏輯空間，及訪問(wèn)系統(tǒng)內(nèi)核對(duì)象(會(huì)話(huà))指向的 pagefile 里面的虛擬文件。無(wú)論是共享堆，還是共享文件，并發(fā)訪問(wèn)都有可能導(dǎo)致數(shù)據(jù)不一致。我們就這個(gè)問(wèn)題簡(jiǎn)單討論一下，該怎樣確保線程/進(jìn)程調(diào)用的有序性及數(shù)據(jù)的一致性。

2.1 線程同步

.NET 框架和 C# 提供了方便直觀的線程同步方法，即 monitor 類(lèi)和 lock 語(yǔ)句(本文將不會(huì)討論 .NET 框架的互斥量)。對(duì)于線程同步，雖然本文提供了其他方法，我們還是推薦使用 lock 語(yǔ)句。

1. void Work1() 
2. { 
3. NonCriticalSection1(); 
4. Monitor.Enter(this); 
5. try 
6. { 
7. CriticalSection(); 
8. } 
9. finally 
10. { 
11. Monitor.Exit(this); 
12. } 
13. NonCriticalSection2(); 
14. } 
15.  
16. void Work2() 
17. { 
18. NonCriticalSection1(); 
19. lock(this) 
20. { 
21. CriticalSection(); 
22. } 
23. NonCriticalSection2(); 
24. } 

Work1 和 Work2 是等價(jià)的。在C#里面，很多人喜歡第二個(gè)方法，因?yàn)樗蹋也蝗菀壮鲥e(cuò)。

2.2 跨線程信號(hào)量

信號(hào)量是經(jīng)典的同步基本概念之一(由 Edsger Dijkstra 引入)。信號(hào)量是指一個(gè)有計(jì)數(shù)器及兩個(gè)操作的對(duì)象。它的兩個(gè)操作是：獲取(也叫P或者等待)，釋放(也叫V或者收到信號(hào))。信號(hào)量在獲取操作時(shí)如果計(jì)數(shù)器為0則阻塞，否則將計(jì)數(shù)器減一;在釋放時(shí)將計(jì)數(shù)器加一，且不會(huì)阻塞。雖然信號(hào)量的原理很簡(jiǎn)單，但是實(shí)現(xiàn)起來(lái)有點(diǎn)麻煩。好在，內(nèi)建的 monitor 類(lèi)有阻塞特性，可以用來(lái)實(shí)現(xiàn)信號(hào)量。

1. public sealed class ThreadSemaphore : ISemaphore 
2. { 
3. private int counter; 
4. private readonly int max; 
5.  
6. public ThreadSemaphore() : this(0, int.Max) {} 
7. public ThreadSemaphore(int initial) : this(initial, int.Max) {} 
8. public ThreadSemaphore(int initial, int max) 
9. { 
10. this.counter = Math.Min(initial,max); 
11. this.max = max; 
12. } 
13.  
14. public void Acquire() 
15. { 
16. lock(this) 
17. { 
18. counter--; 
19. if(counter < 0 && !Monitor.Wait(this)) 
20. throw new SemaphoreFailedException(); 
21. } 
22. } 
23.  
24. public void Acquire(TimeSpan timeout) 
25. { 
26. lock(this) 
27. { 
28. counter--; 
29. if(counter < 0 && !Monitor.Wait(this,timeout)) 
30. throw new SemaphoreFailedException(); 
31. } 
32. } 
33.  
34. public void Release() 
35. { 
36. lock(this) 
37. { 
38. if(counter >= max) 
39. throw new SemaphoreFailedException(); 
40. if(counter < 0) 
41. Monitor.Pulse(this); 
42. counter++; 
43. } 
44. } 
45. } 

信號(hào)量在復(fù)雜的阻塞情景下更加有用，例如我們后面將要討論的通道(channel)。你也可以使用信號(hào)量來(lái)實(shí)現(xiàn)臨界區(qū)的排他性(如下面的 Work3)，但是我還是推薦使用內(nèi)建的 lock 語(yǔ)句，像上面的 Work2 那樣。

請(qǐng)注意：如果使用不當(dāng)，信號(hào)量也是有潛在危險(xiǎn)的。正確的做法是：當(dāng)獲取信號(hào)量失敗時(shí)，千萬(wàn)不要再調(diào)用釋放操作;當(dāng)獲取成功時(shí)，無(wú)論發(fā)生了什么錯(cuò)誤，都要記得釋放信號(hào)量。遵循這樣的原則，你的同步才是正確的。Work3 中的 finally 語(yǔ)句就是為了保證正確釋放信號(hào)量。注意：獲取信號(hào)量( s.Acquire() )的操作必須放到 try 語(yǔ)句的外面，只有這樣，當(dāng)獲取失敗時(shí)才不會(huì)調(diào)用釋放操作。

1. ThreadSemaphore s = new ThreadSemaphore(1); 
2. void Work3() 
3. { 
4. NonCriticalSection1(); 
5. s.Acquire(); 
6. try 
7. { 
8. CriticalSection(); 
9. } 
10. finally 
11. { 
12. s.Release(); 
13. } 
14. NonCriticalSection2(); 
15. } 

2.3 跨進(jìn)程信號(hào)量

為了協(xié)調(diào)不同進(jìn)程訪問(wèn)同一資源，我們需要用到上面討論過(guò)的概念。很不幸，.NET 中的 monitor 類(lèi)不可以跨進(jìn)程使用。但是，win32 API提供的內(nèi)核信號(hào)量對(duì)象可以用來(lái)實(shí)現(xiàn)跨進(jìn)程同步。 Robin Galloway-Lunn 介紹了怎樣將 win32 的信號(hào)量映射到 .NET 中(見(jiàn) Using Win32 Semaphores in C# )。我們的實(shí)現(xiàn)也類(lèi)似：

1. [DllImport("kernel32",EntryPoint="CreateSemaphore", 
2. SetLastError=true,CharSet=CharSet.Unicode)] 
3. internal static extern uint CreateSemaphore( 
4. SecurityAttributes auth, int initialCount, 
5. int maximumCount, string name); 
6.  
7. [DllImport("kernel32",EntryPoint="WaitForSingleObject", 
8. SetLastError=true,CharSet=CharSet.Unicode)] 
9. internal static extern uint WaitForSingleObject( 
10. uint hHandle, uint dwMilliseconds); 
11.  
12. [DllImport("kernel32",EntryPoint="ReleaseSemaphore", 
13. SetLastError=true,CharSet=CharSet.Unicode)] 
14. [return : MarshalAs( UnmanagedType.VariantBool )] 
15. internal static extern bool ReleaseSemaphore( 
16. uint hHandle, int lReleaseCount, out int lpPreviousCount); 
17.  
18. [DllImport("kernel32",EntryPoint="CloseHandle",SetLastError=true, 
19. CharSet=CharSet.Unicode)] 
20. [return : MarshalAs( UnmanagedType.VariantBool )] 
21. internal static extern bool CloseHandle(uint hHandle); 
22.  
23. public class ProcessSemaphore : ISemaphore, IDisposable 
24. { 
25. private uint handle; 
26. private readonly uint interruptReactionTime; 
27.  
28. public ProcessSemaphore(string name) : this( 
29. name,0,int.MaxValue,500) {} 
30. public ProcessSemaphore(string name, int initial) : this( 
31. name,initial,int.MaxValue,500) {} 
32. public ProcessSemaphore(string name, int initial, 
33. int max, int interruptReactionTime) 
34. {  
35. this.interruptReactionTime = (uint)interruptReactionTime; 
36. this.handle = NTKernel.CreateSemaphore(null, initial, max, name); 
37. if(handle == 0) 
38. throw new SemaphoreFailedException(); 
39. } 
40.  
41. public void Acquire() 
42. { 
43. while(true) 
44. { //looped 0.5s timeout to make NT-blocked threads interruptable. 
45. uint res = NTKernel.WaitForSingleObject(handle, 
46. interruptReactionTime); 
47. try {System.Threading.Thread.Sleep(0);} 
48. catch(System.Threading.ThreadInterruptedException e) 
49. { 
50. if(res == 0) 
51. { //Rollback 
52. int previousCount; 
53. NTKernel.ReleaseSemaphore(handle,1,out previousCount); 
54. } 
55. throw e; 
56. } 
57. if(res == 0) 
58. return; 
59. if(res != 258) 
60. throw new SemaphoreFailedException(); 
61. } 
62. } 
63.  
64. public void Acquire(TimeSpan timeout) 
65. { 
66. uint milliseconds = (uint)timeout.TotalMilliseconds; 
67. if(NTKernel.WaitForSingleObject(handle, milliseconds) != 0) 
68. throw new SemaphoreFailedException();  
69. } 
70.  
71. public void Release() 
72. { 
73. int previousCount; 
74. if(!NTKernel.ReleaseSemaphore(handle, 1, out previousCount)) 
75. throw new SemaphoreFailedException();  
76. } 
77.  
78. #region IDisposable Member 
79. public void Dispose() 
80. { 
81. if(handle != 0) 
82. { 
83. if(NTKernel.CloseHandle(handle)) 
84. handle = 0; 
85. } 
86. } 
87. #endregion 
88. } 

有一點(diǎn)很重要：win32中的信號(hào)量是可以命名的。這允許其他進(jìn)程通過(guò)名字來(lái)創(chuàng)建相應(yīng)信號(hào)量的句柄。為了讓阻塞線程可以中斷，我們使用了一個(gè)(不好)的替代方法：使用超時(shí)和 Sleep(0)。我們需要中斷來(lái)安全關(guān)閉線程。更好的做法是：確定沒(méi)有線程阻塞之后才釋放信號(hào)量，這樣程序才可以完全釋放資源并正確退出。

你可能也注意到了：跨線程和跨進(jìn)程的信號(hào)量都使用了相同的接口。所有相關(guān)的類(lèi)都使用了這種模式，以實(shí)現(xiàn)上面背景介紹中提到的封閉性。需要注意：出于性能考慮，你不應(yīng)該將跨進(jìn)程的信號(hào)量用到跨線程的場(chǎng)景，也不應(yīng)該將跨線程的實(shí)現(xiàn)用到單線程的場(chǎng)景。

3. 跨進(jìn)程共享內(nèi)存：內(nèi)存映射文件

我們已經(jīng)實(shí)現(xiàn)了跨線程和跨進(jìn)程的共享資源訪問(wèn)同步。但是傳遞/接收消息還需要共享資源。對(duì)于線程來(lái)說(shuō)，只需要聲明一個(gè)類(lèi)成員變量就可以了。但是對(duì)于跨進(jìn)程來(lái)說(shuō)，我們需要使用到 win32 API 提供的內(nèi)存映射文件(Memory Mapped Files，簡(jiǎn)稱(chēng)MMF)。使用 MMF和使用 win32 信號(hào)量差不多。我們需要先調(diào)用 CreateFileMapping 方法來(lái)創(chuàng)建一個(gè)內(nèi)存映射文件的句柄：

1. [DllImport("Kernel32.dll",EntryPoint="CreateFileMapping", 
2. SetLastError=true,CharSet=CharSet.Unicode)] 
3. internal static extern IntPtr CreateFileMapping(uint hFile, 
4. SecurityAttributes lpAttributes, uint flProtect, 
5. uint dwMaximumSizeHigh, uint dwMaximumSizeLow, string lpName); 
6.  
7. [DllImport("Kernel32.dll",EntryPoint="MapViewOfFile", 
8. SetLastError=true,CharSet=CharSet.Unicode)] 
9. internal static extern IntPtr MapViewOfFile(IntPtr hFileMappingObject, 
10. uint dwDesiredAccess, uint dwFileOffsetHigh, 
11. uint dwFileOffsetLow, uint dwNumberOfBytesToMap); 
12.  
13. [DllImport("Kernel32.dll",EntryPoint="UnmapViewOfFile", 
14. SetLastError=true,CharSet=CharSet.Unicode)] 
15. [return : MarshalAs( UnmanagedType.VariantBool )] 
16. internal static extern bool UnmapViewOfFile(IntPtr lpBaseAddress); 
17.  
18. public static MemoryMappedFile CreateFile(string name, 
19. FileAccess access, int size) 
20. { 
21. if(size < 0) 
22. throw new ArgumentException("Size must not be negative","size"); 
23.  
24. IntPtr fileMapping = NTKernel.CreateFileMapping(0xFFFFFFFFu,null, 
25. (uint)access,0,(uint)size,name); 
26. if(fileMapping == IntPtr.Zero) 
27. throw new MemoryMappingFailedException(); 
28.  
29. return new MemoryMappedFile(fileMapping,size,access); 
30. } 

我們希望直接使用 pagefile 中的虛擬文件，所以我們用 -1(0xFFFFFFFF) 來(lái)作為文件句柄來(lái)創(chuàng)建我們的內(nèi)存映射文件句柄。我們也指定了必填的文件大小，以及相應(yīng)的名稱(chēng)。這樣其他進(jìn)程就可以通過(guò)這個(gè)名稱(chēng)來(lái)同時(shí)訪問(wèn)該映射文件。創(chuàng)建了內(nèi)存映射文件后，我們就可以映射這個(gè)文件不同的部分(通過(guò)偏移量和字節(jié)大小來(lái)指定)到我們的進(jìn)程地址空間。我們通過(guò) MapViewOfFile 系統(tǒng)方法來(lái)指定：

1. public MemoryMappedFileView CreateView(int offset, int size, 
2. MemoryMappedFileView.ViewAccess access) 
3. { 
4. if(this.access == FileAccess.ReadOnly && access == 
5. MemoryMappedFileView.ViewAccess.ReadWrite) 
6. throw new ArgumentException( 
7. "Only read access to views allowed on files without write access", 
8. "access"); 
9. if(offset < 0) 
10. throw new ArgumentException("Offset must not be negative","size"); 
11. if(size < 0) 
12. throw new ArgumentException("Size must not be negative","size"); 
13. IntPtr mappedView = NTKernel.MapViewOfFile(fileMapping, 
14. (uint)access,0,(uint)offset,(uint)size); 
15. return new MemoryMappedFileView(mappedView,size,access); 
16. } 

在不安全的代碼中，我們可以將返回的指針強(qiáng)制轉(zhuǎn)換成我們指定的類(lèi)型。盡管如此，我們不希望有不安全的代碼存在，所以我們使用 Marshal 類(lèi)來(lái)從中讀寫(xiě)我們的數(shù)據(jù)。偏移量參數(shù)是用來(lái)從哪里開(kāi)始讀寫(xiě)數(shù)據(jù)，相對(duì)于指定的映射視圖的地址。

1. public byte ReadByte(int offset) 
2. { 
3. return Marshal.ReadByte(mappedView,offset); 
4. } 
5. public void WriteByte(byte data, int offset) 
6. { 
7. Marshal.WriteByte(mappedView,offset,data); 
8. } 
9.  
10. public int ReadInt32(int offset) 
11. { 
12. return Marshal.ReadInt32(mappedView,offset); 
13. } 
14. public void WriteInt32(int data, int offset) 
15. { 
16. Marshal.WriteInt32(mappedView,offset,data); 
17. } 
18.  
19. public void ReadBytes(byte[] data, int offset) 
20. { 
21. for(int i=0;i<data.Length;i++) 
22. data[i] = Marshal.ReadByte(mappedView,offset+i); 
23. } 
24. public void WriteBytes(byte[] data, int offset) 
25. { 
26. for(int i=0;i<data.Length;i++) 
27. Marshal.WriteByte(mappedView,offset+i,data[i]); 
28. } 

但是，我們希望讀寫(xiě)整個(gè)對(duì)象樹(shù)到文件中，所以我們需要支持自動(dòng)進(jìn)行序列化和反序列化的方法。

1. public object ReadDeserialize(int offset, int length) 
2. { 
3. byte[] binaryData = new byte[length]; 
4. ReadBytes(binaryData,offset); 
5. System.Runtime.Serialization.Formatters.Binary.BinaryFormatter formatter 
6. = new System.Runtime.Serialization.Formatters.Binary.BinaryFormatter(); 
7. System.IO.MemoryStream ms = new System.IO.MemoryStream( 
8. binaryData,0,length,true,true); 
9. object data = formatter.Deserialize(ms); 
10. ms.Close(); 
11. return data; 
12. } 
13. public void WriteSerialize(object data, int offset, int length) 
14. { 
15. System.Runtime.Serialization.Formatters.Binary.BinaryFormatter formatter 
16. = new System.Runtime.Serialization.Formatters.Binary.BinaryFormatter(); 
17. byte[] binaryData = new byte[length]; 
18. System.IO.MemoryStream ms = new System.IO.MemoryStream( 
19. binaryData,0,length,true,true); 
20. formatter.Serialize(ms,data); 
21. ms.Flush(); 
22. ms.Close(); 
23. WriteBytes(binaryData,offset); 
24. } 

請(qǐng)注意：對(duì)象序列化之后的大小不應(yīng)該超過(guò)映射視圖的大小。序列化之后的大小總是比對(duì)象本身占用的內(nèi)存要大的。我沒(méi)有試過(guò)直接將對(duì)象內(nèi)存流綁定到映射視圖，那樣做應(yīng)該也可以，甚至可能帶來(lái)少量的性能提升。

4. 信箱：在線程/進(jìn)程間傳遞消息

這里的信箱與 Email 及 NT 中的郵件槽(Mailslots)無(wú)關(guān)。它是一個(gè)只能保留一個(gè)對(duì)象的安全共享內(nèi)存結(jié)構(gòu)。信箱的內(nèi)容通過(guò)一個(gè)屬性來(lái)讀寫(xiě)。如果信箱內(nèi)容為空，試圖讀取該信箱的線程將會(huì)阻塞，直到另一個(gè)線程往其中寫(xiě)內(nèi)容。如果信箱已經(jīng)有了內(nèi)容，當(dāng)一個(gè)線程試圖往其中寫(xiě)內(nèi)容時(shí)將被阻塞，直到另一個(gè)線程將信箱內(nèi)容讀取出去。信箱的內(nèi)容只能被讀取一次，它的引用在讀取后自動(dòng)被刪除。基于上面的代碼，我們已經(jīng)可以實(shí)現(xiàn)信箱了。

4.1 跨線程的信箱

我們可以使用兩個(gè)信號(hào)量來(lái)實(shí)現(xiàn)一個(gè)信箱：一個(gè)信號(hào)量在信箱內(nèi)容為空時(shí)觸發(fā)，另一個(gè)在信箱有內(nèi)容時(shí)觸發(fā)。在讀取內(nèi)容之前，線程先等待信箱已經(jīng)填充了內(nèi)容，讀取之后觸發(fā)空信號(hào)量。在寫(xiě)入內(nèi)容之前，線程先等待信箱內(nèi)容清空，寫(xiě)入之后觸發(fā)滿(mǎn)信號(hào)量。注意：空信號(hào)量在一開(kāi)始時(shí)就被觸發(fā)了。

1. public sealed class ThreadMailBox : IMailBox 
2. { 
3. private object content; 
4. private ThreadSemaphore empty, full; 
5.  
6. public ThreadMailBox() 
7. { 
8. empty = new ThreadSemaphore(1,1); 
9. full = new ThreadSemaphore(0,1); 
10. } 
11.  
12. public object Content 
13. { 
14. get 
15. { 
16. full.Acquire(); 
17. object item = content; 
18. empty.Release(); 
19. return item; 
20. } 
21. set 
22. { 
23. empty.Acquire(); 
24. content = value; 
25. full.Release(); 
26. } 
27. } 
28. } 

4.2 跨進(jìn)程信箱

跨進(jìn)程信箱與跨線程信箱的實(shí)現(xiàn)基本上一樣簡(jiǎn)單。不同的是我們使用兩個(gè)跨進(jìn)程的信號(hào)量，并且我們使用內(nèi)存映射文件來(lái)代替類(lèi)成員變量。由于序列化可能會(huì)失敗，我們使用了一小段異常處理來(lái)回滾信箱的狀態(tài)。失敗的原因有很多(無(wú)效句柄，拒絕訪問(wèn)，文件大小問(wèn)題，Serializable屬性缺失等等)。

1. public sealed class ProcessMailBox : IMailBox, IDisposable 
2. { 
3. private MemoryMappedFile file; 
4. private MemoryMappedFileView view; 
5. private ProcessSemaphore empty, full; 
6.  
7. public ProcessMailBox(string name,int size) 
8. { 
9. empty = new ProcessSemaphore(name+".EmptySemaphore.MailBox",1,1); 
10. full = new ProcessSemaphore(name+".FullSemaphore.MailBox",0,1); 
11. file = MemoryMappedFile.CreateFile(name+".MemoryMappedFile.MailBox", 
12. MemoryMappedFile.FileAccess.ReadWrite,size); 
13. view = file.CreateView(0,size, 
14. MemoryMappedFileView.ViewAccess.ReadWrite); 
15. } 
16.  
17. public object Content 
18. { 
19. get 
20. { 
21. full.Acquire(); 
22. object item; 
23. try {item = view.ReadDeserialize();} 
24. catch(Exception e) 
25. { //Rollback 
26. full.Release(); 
27. throw e; 
28. } 
29. empty.Release(); 
30. return item; 
31. } 
32.  
33. set 
34. { 
35. empty.Acquire(); 
36. try {view.WriteSerialize(value);} 
37. catch(Exception e) 
38. { //Rollback 
39. empty.Release(); 
40. throw e; 
41. } 
42. full.Release(); 
43. } 
44. } 
45.  
46. #region IDisposable Member 
47. public void Dispose() 
48. { 
49. view.Dispose(); 
50. file.Dispose(); 
51. empty.Dispose(); 
52. full.Dispose(); 
53. } 
54. #endregion 
55. } 

到這里我們已經(jīng)實(shí)現(xiàn)了跨進(jìn)程消息傳遞(IPC)所需要的組件。你可能需要再回頭本文開(kāi)頭的那個(gè)例子，看看 ProcessMailBox 應(yīng)該如何使用。

5.通道：基于隊(duì)列的消息傳遞

信箱最大的限制是它們每次只能保存一個(gè)對(duì)象。如果一系列線程(使用同一個(gè)信箱)中的一個(gè)線程需要比較長(zhǎng)的時(shí)間來(lái)處理特定的命令，那么整個(gè)系列都會(huì)阻塞。通常我們會(huì)使用緩沖的消息通道來(lái)處理，這樣你可以在方便的時(shí)候從中讀取消息，而不會(huì)阻塞消息發(fā)送者。這種緩沖通過(guò)通道來(lái)實(shí)現(xiàn)，這里的通道比信箱要復(fù)雜一些。同樣，我們將分別從線程和進(jìn)程級(jí)別來(lái)討論通道的實(shí)現(xiàn)。

5.1 可靠性

信箱和通道的另一個(gè)重要的不同是：通道擁有可靠性。例如：自動(dòng)將發(fā)送失敗(可能由于線程等待鎖的過(guò)程中被中斷)的消息轉(zhuǎn)存到一個(gè)內(nèi)置的容器中。這意味著處理通道的線程可以安全地停止，同時(shí)不會(huì)丟失隊(duì)列中的消息。這通過(guò)兩個(gè)抽象類(lèi)來(lái)實(shí)現(xiàn)， ThreadReliability 和 ProcessReliability。每個(gè)通道的實(shí)現(xiàn)類(lèi)都繼承其中的一個(gè)類(lèi)。

5.2 跨線程的通道

跨線程的通道基于信箱來(lái)實(shí)現(xiàn)，但是使用一個(gè)同步的隊(duì)列來(lái)作為消息緩沖而不是一個(gè)變量。得益于信號(hào)量，通道在空隊(duì)列時(shí)阻塞接收線程，在隊(duì)列滿(mǎn)時(shí)阻塞發(fā)送線程。這樣你就不會(huì)碰到由入隊(duì)/出隊(duì)引發(fā)的錯(cuò)誤。為了實(shí)現(xiàn)這個(gè)效果，我們用隊(duì)列大小來(lái)初始化空信號(hào)量，用0來(lái)初始化滿(mǎn)信號(hào)量。如果某個(gè)發(fā)送線程在等待入隊(duì)的時(shí)候被中斷，我們將消息復(fù)制到內(nèi)置容器中，并將異常往外面拋。在接收操作中，我們不需要做異常處理，因?yàn)榧词咕€程被中斷你也不會(huì)丟失任何消息。注意：線程只有在阻塞狀態(tài)才能被中斷，就像調(diào)用信號(hào)量的獲取操作(Aquire)方法時(shí)。

1. public sealed class ThreadChannel : ThreadReliability, IChannel 
2. { 
3. private Queue queue; 
4. private ThreadSemaphore empty, full; 
5.  
6. public ThreadChannel(int size) 
7. { 
8. queue = Queue.Synchronized(new Queue(size)); 
9. empty = new ThreadSemaphore(size,size); 
10. full = new ThreadSemaphore(0,size); 
11. } 
12.  
13. public void Send(object item) 
14. { 
15. try {empty.Acquire();} 
16. catch(System.Threading.ThreadInterruptedException e) 
17. { 
18. DumpItem(item); 
19. throw e; 
20. } 
21. queue.Enqueue(item); 
22. full.Release(); 
23. } 
24.  
25. public void Send(object item, TimeSpan timeout) 
26. { 
27. try {empty.Acquire(timeout);} 
28. ... 
29. } 
30.  
31. public object Receive() 
32. { 
33. full.Acquire(); 
34. object item = queue.Dequeue(); 
35. empty.Release(); 
36. return item; 
37. } 
38.  
39. public object Receive(TimeSpan timeout) 
40. { 
41. full.Acquire(timeout); 
42. ... 
43. } 
44.  
45. protected override void DumpStructure() 
46. { 
47. lock(queue.SyncRoot) 
48. { 
49. foreach(object item in queue) 
50. DumpItem(item); 
51. queue.Clear(); 
52. } 
53. } 
54. } 

5.3 跨進(jìn)程通道

實(shí)現(xiàn)跨進(jìn)程通道有點(diǎn)麻煩，因?yàn)槟阈枰紫忍峁┮粋€(gè)跨進(jìn)程的緩沖區(qū)。一個(gè)可能的解決方法是使用跨進(jìn)程信箱并根據(jù)需要將接收/發(fā)送方法加入隊(duì)列。為了避免這種方案的幾個(gè)缺點(diǎn)，我們將直接使用內(nèi)存映射文件來(lái)實(shí)現(xiàn)一個(gè)隊(duì)列。MemoryMappedArray 類(lèi)將內(nèi)存映射文件分成幾部分，可以直接使用數(shù)組索引來(lái)訪問(wèn)。 MemoryMappedQueue 類(lèi)，為這個(gè)數(shù)組提供了一個(gè)經(jīng)典的環(huán)(更多細(xì)節(jié)請(qǐng)查看附件中的代碼)。為了支持直接以 byte/integer 類(lèi)型訪問(wèn)數(shù)據(jù)并同時(shí)支持二進(jìn)制序列化，調(diào)用方需要先調(diào)用入隊(duì)(Enqueue)/出隊(duì)(Dequeue)操作，然后根據(jù)需要使用讀寫(xiě)方法(隊(duì)列會(huì)自動(dòng)將數(shù)據(jù)放到正確的位置)。這兩個(gè)類(lèi)都不是線程和進(jìn)程安全的，所以我們需要使用跨進(jìn)程的信號(hào)量來(lái)模擬互斥量(也可以使用 win32 互斥量)，以此實(shí)現(xiàn)相互間的互斥訪問(wèn)。除了這兩個(gè)類(lèi)，跨進(jìn)程的通道基本上和跨線程信箱一樣。同樣，我們也需要在 Send() 中處理線程中斷及序列化可能失敗的問(wèn)題。

1. public sealed class ProcessChannel : ProcessReliability, IChannel, IDisposable 
2. { 
3. private MemoryMappedFile file; 
4. private MemoryMappedFileView view; 
5. private MemoryMappedQueue queue; 
6. private ProcessSemaphore empty, full, mutex; 
7.  
8. public ProcessChannel( int size, string name, int maxBytesPerEntry) 
9. { 
10. int fileSize = 64+size*maxBytesPerEntry; 
11.  
12. empty = new ProcessSemaphore(name+".EmptySemaphore.Channel",size,size); 
13. full = new ProcessSemaphore(name+".FullSemaphore.Channel",0,size); 
14. mutex = new ProcessSemaphore(name+".MutexSemaphore.Channel",1,1); 
15. file = MemoryMappedFile.CreateFile(name+".MemoryMappedFile.Channel", 
16. MemoryMappedFile.FileAccess.ReadWrite,fileSize); 
17. view = file.CreateView(0,fileSize, 
18. MemoryMappedFileView.ViewAccess.ReadWrite); 
19. queue = new MemoryMappedQueue(view,size,maxBytesPerEntry,true,0); 
20. if(queue.Length < size || queue.BytesPerEntry < maxBytesPerEntry) 
21. throw new MemoryMappedArrayFailedException(); 
22. } 
23.  
24. public void Send(object item) 
25. { 
26. try {empty.Acquire();} 
27. catch(System.Threading.ThreadInterruptedException e) 
28. { 
29. DumpItemSynchronized(item); 
30. throw e; 
31. } 
32. try {mutex.Acquire();} 
33. catch(System.Threading.ThreadInterruptedException e) 
34. { 
35. DumpItemSynchronized(item); 
36. empty.Release(); 
37. throw e; 
38. } 
39. queue.Enqueue(); 
40. try {queue.WriteSerialize(item,0);} 
41. catch(Exception e) 
42. { 
43. queue.RollbackEnqueue(); 
44. mutex.Release(); 
45. empty.Release(); 
46. throw e; 
47. } 
48. mutex.Release(); 
49. full.Release(); 
50. } 
51.  
52. public void Send(object item, TimeSpan timeout) 
53. { 
54. try {empty.Acquire(timeout);} 
55. ... 
56. } 
57.  
58. public object Receive() 
59. { 
60. full.Acquire(); 
61. mutex.Acquire(); 
62. object item; 
63. queue.Dequeue(); 
64. try {item = queue.ReadDeserialize(0);} 
65. catch(Exception e) 
66. { 
67. queue.RollbackDequeue(); 
68. mutex.Release(); 
69. full.Release(); 
70. throw e; 
71. } 
72. mutex.Release(); 
73. empty.Release(); 
74. return item; 
75. } 
76.  
77. public object Receive(TimeSpan timeout) 
78. { 
79. full.Acquire(timeout); 
80. ... 
81. } 
82.  
83. protected override void DumpStructure() 
84. { 
85. mutex.Acquire(); 
86. byte[][] dmp = queue.DumpClearAll(); 
87. for(int i=0;i<dmp.Length;i++) 
88. DumpItemSynchronized(dmp[i]); 
89. mutex.Release(); 
90. } 
91.  
92. #region IDisposable Member 
93. public void Dispose() 
94. { 
95. view.Dispose(); 
96. file.Dispose(); 
97. empty.Dispose(); 
98. full.Dispose(); 
99. mutex.Dispose(); 
100. } 
101. #endregion 
102. } 

6. 消息路由

我們目前已經(jīng)實(shí)現(xiàn)了線程和進(jìn)程同步及消息傳遞機(jī)制(使用信箱和通道)。當(dāng)你使用阻塞隊(duì)列的時(shí)候，有可能會(huì)遇到這樣的問(wèn)題：你需要在一個(gè)線程中同時(shí)監(jiān)聽(tīng)多個(gè)隊(duì)列。為了解決這樣的問(wèn)題，我們提供了一些小型的類(lèi)：通道轉(zhuǎn)發(fā)器，多用復(fù)用器，多路復(fù)用解碼器和通道事件網(wǎng)關(guān)。你也可以通過(guò)簡(jiǎn)單的 IRunnable 模式來(lái)實(shí)現(xiàn)類(lèi)似的通道處理器。IRunnable模式由兩個(gè)抽象類(lèi)SingleRunnable和 MultiRunnable 來(lái)提供(具體細(xì)節(jié)請(qǐng)參考附件中的代碼)。

6.1 通道轉(zhuǎn)發(fā)器

通道轉(zhuǎn)發(fā)器僅僅監(jiān)聽(tīng)一個(gè)通道，然后將收到的消息轉(zhuǎn)發(fā)到另一個(gè)通道。如果有必要，轉(zhuǎn)發(fā)器可以將每個(gè)收到的消息放到一個(gè)信封中，并加上一個(gè)數(shù)字標(biāo)記，然后再轉(zhuǎn)發(fā)出去(下面的多路利用器使用了這個(gè)特性)。

1. public class ChannelForwarder : SingleRunnable 
2. { 
3. private IChannel source, target; 
4. private readonly int envelope; 
5.  
6. public ChannelForwarder(IChannel source, 
7. IChannel target, bool autoStart, bool waitOnStop) 
8. : base(true,autoStart,waitOnStop) 
9. { 
10. this.source = source; 
11. this.target = target; 
12. this.envelope = -1; 
13. } 
14. public ChannelForwarder(IChannel source, IChannel target, 
15. int envelope, bool autoStart, bool waitOnStop) 
16. : base(true,autoStart,waitOnStop) 
17. { 
18. this.source = source; 
19. this.target = target; 
20. this.envelope = envelope; 
21. } 
22.  
23. protected override void Run() 
24. { //NOTE: IChannel.Send is interrupt save and 
25. //automatically dumps the argument. 
26. if(envelope == -1) 
27. while(running) 
28. target.Send(source.Receive()); 
29. else 
30. { 
31. MessageEnvelope env; 
32. env.ID = envelope; 
33. while(running) 
34. { 
35. env.Message = source.Receive(); 
36. target.Send(env); 
37. } 
38. } 
39. } 
40. } 

6.2 通道多路復(fù)用器和通道復(fù)用解碼器

通道多路復(fù)用器監(jiān)聽(tīng)多個(gè)來(lái)源的通道并將接收到的消息(消息使用信封來(lái)標(biāo)記來(lái)源消息)轉(zhuǎn)發(fā)到一個(gè)公共的輸出通道。這樣就可以一次性地監(jiān)聽(tīng)多個(gè)通道。復(fù)用解碼器則是監(jiān)聽(tīng)一個(gè)公共的輸出通道，然后根據(jù)信封將消息轉(zhuǎn)發(fā)到某個(gè)指定的輸出通道。

1. public class ChannelMultiplexer : MultiRunnable 
2. { 
3. private ChannelForwarder[] forwarders; 
4.  
5. public ChannelMultiplexer(IChannel[] channels, int[] ids, 
6. IChannel output, bool autoStart, bool waitOnStop) 
7. { 
8. int count = channels.Length; 
9. if(count != ids.Length) 
10. throw new ArgumentException("Channel and ID count mismatch.","ids"); 
11.  
12. forwarders = new ChannelForwarder[count]; 
13. for(int i=0;i<count;i++) 
14. forwarders[i] = new ChannelForwarder(channels[i], 
15. output,ids[i],autoStart,waitOnStop); 
16.  
17. SetRunnables((SingleRunnable[])forwarders); 
18. } 
19. } 
20.  
21. public class ChannelDemultiplexer : SingleRunnable 
22. { 
23. private HybridDictionary dictionary; 
24. private IChannel input; 
25.  
26. public ChannelDemultiplexer(IChannel[] channels, int[] ids, 
27. IChannel input, bool autoStart, bool waitOnStop) 
28. : base(true,autoStart,waitOnStop) 
29. { 
30. this.input = input; 
31.  
32. int count = channels.Length; 
33. if(count != ids.Length) 
34. throw new ArgumentException("Channel and ID count mismatch.","ids"); 
35.  
36. dictionary = new HybridDictionary(count,true); 
37. for(int i=0;i<count;i++) 
38. dictionary.add(ids[i],channels[i]); 
39. } 
40.  
41. protected override void Run() 
42. { //NOTE: IChannel.Send is interrupt save and 
43. //automatically dumps the argument. 
44. while(running) 
45. { 
46. MessageEnvelope env = (MessageEnvelope)input.Receive(); 
47. IChannel channel = (IChannel)dictionary[env.ID]; 
48. channel.send(env.Message); 
49. } 
50. } 
51. } 

6.3 通道事件網(wǎng)關(guān)

通道事件網(wǎng)關(guān)監(jiān)聽(tīng)指定的通道，在接收到消息時(shí)觸發(fā)一個(gè)事件。這個(gè)類(lèi)對(duì)于基于事件的程序(例如GUI程序)很有用，或者在使用系統(tǒng)線程池(ThreadPool)來(lái)初始化輕量的線程。需要注意的是：使用 WinForms 的程序中你不能在事件處理方法中直接訪問(wèn)UI控件，只能調(diào)用Invoke 方法。因?yàn)槭录幚矸椒ㄊ怯墒录W(wǎng)關(guān)線程調(diào)用的，而不是UI線程。

1. public class ChannelEventGateway : SingleRunnable 
2. { 
3. private IChannel source; 
4. public event MessageReceivedEventHandler MessageReceived; 
5.  
6. public ChannelEventGateway(IChannel source, bool autoStart, 
7. bool waitOnStop) : base(true,autoStart,waitOnStop) 
8. { 
9. this.source = source; 
10. } 
11.  
12. protected override void Run() 
13. { 
14. while(running) 
15. { 
16. object c = source.Receive(); 
17. MessageReceivedEventHandler handler = MessageReceived; 
18. if(handler != null) 
19. handler(this,new MessageReceivedEventArgs(c)); 
20. } 
21. } 
22. } 

7. 比薩外賣(mài)店的例子

萬(wàn)事俱備，只欠東風(fēng)。我們已經(jīng)討論了這個(gè)同步及消息傳遞框架中的大部分重要的結(jié)構(gòu)和技術(shù)(本文沒(méi)有討論框架中的其他類(lèi)如Rendezvous及Barrier)。就像開(kāi)頭一樣，我們用一個(gè)例子來(lái)結(jié)束這篇文章。這次我們用一個(gè)小型比薩外賣(mài)店來(lái)做演示。下圖展示了這個(gè)例子：四個(gè)并行進(jìn)程相互之間進(jìn)行通訊。圖中展示了消息(數(shù)據(jù))是如何使用跨進(jìn)程通道在四個(gè)進(jìn)程中流動(dòng)的，且在每個(gè)進(jìn)程中使用了性能更佳的跨線程通道和信箱。

一開(kāi)始，一個(gè)顧客點(diǎn)了一個(gè)比薩和一些飲料。他調(diào)用了顧客(customer)接口的方法，向顧客訂單(CustomerOrders)通道發(fā)送了一個(gè)下單(Order)消息。接單員，在顧客下單后，發(fā)送了兩條配餐指令(分別對(duì)應(yīng)比薩和飲料)到廚師指令(CookInstruction)通道。同時(shí)他通過(guò)收銀(CashierOrder)通道將訂單轉(zhuǎn)發(fā)給收銀臺(tái)。收銀臺(tái)從價(jià)格中心獲取總價(jià)并將票據(jù)發(fā)給顧客，希望能提高收銀的速度 。與此同時(shí)，廚師將根據(jù)配餐指令將餐配好之后交給打包員工。打包員工處理好之后，等待顧客付款，然后將外賣(mài)遞給顧客。

為了運(yùn)行這個(gè)例子，打開(kāi)4個(gè)終端(cmd.exe)，用 "PizzaDemo.exe cook" 啟動(dòng)多個(gè)廚師進(jìn)程(多少個(gè)都可以)，用 "PizzaDemo.exe backend" 啟動(dòng)后端進(jìn)程，用 "PizzaDemo.exe facade" 啟動(dòng)顧客接口門(mén)面(用你的程序名稱(chēng)來(lái)代替 PizzaDemo )。注意：為了模擬真實(shí)情景，某些線程(例如廚師線程)會(huì)隨機(jī)休眠幾秒。按下回車(chē)鍵就會(huì)停止和退出進(jìn)程。如果你在進(jìn)程正在處理數(shù)據(jù)的時(shí)候退出，你將可以在內(nèi)存轉(zhuǎn)存報(bào)告的結(jié)尾看到幾個(gè)未處理的消息。在真實(shí)世界的程序里面，消息一般都會(huì)被轉(zhuǎn)存到磁盤(pán)中，以便下次可以使用。

這個(gè)例子使用了上文中討論過(guò)的幾個(gè)機(jī)制。比如說(shuō)，收銀臺(tái)使用一個(gè)通道復(fù)用器(ChannelMultiplexer)來(lái)監(jiān)聽(tīng)顧客的訂單和支付通道，用了兩個(gè)信箱來(lái)實(shí)現(xiàn)價(jià)格服務(wù)。分發(fā)時(shí)使用了一個(gè)通道事件網(wǎng)關(guān)(ChannelEventGateway)，顧客在食物打包完成之后馬上會(huì)收到通知。你也可以將這些程序注冊(cè)成 Windows NT 服務(wù)運(yùn)行，也可以遠(yuǎn)程登錄后運(yùn)行。

8. 總結(jié)

本文已經(jīng)討論了C#中如何基于服務(wù)的架構(gòu)及實(shí)現(xiàn)跨進(jìn)程同步和通訊。然后，這個(gè)不是唯一的解決方案。例如：在大項(xiàng)目中使用那么多的線程會(huì)引來(lái)嚴(yán)重的問(wèn)題。這個(gè)框架中缺失的是事務(wù)支持及其他的通道/信箱實(shí)現(xiàn)(例如命名管道和TCP sockets)。這個(gè)框架中可能也有許多不足之處。