這篇文章主要介紹了使用C#來編寫一個異步的Socket服務器,通過無阻塞機制來獲取更高的處理效率,需要的朋友可以參考下

介紹

我最近需要為一個.net項目準備一個內部線程通信機制. 項目有多個使用ASP.NET，Windows 表單和控制臺應用程序的服務器和客戶端構成. 考慮到實現的可能性，我下定決心要使用原生的socket，而不是許多.NET中已經提前為我們構建好的組件, 像是所謂的管道, NetTcpClient 還有 Azure 服務總線.

這篇文章中的服務器基于System.Net.Sockets類異步方法. 這些允許你支持大量的socket客戶端, 而一個客戶端的連接是唯一的阻塞機制. 阻塞的時間是可以忽略不記得，所以服務器基本上是在當做一個多線程socket服務器在運作的.

背景

原生的socket在為你提供通信層面的完全控制權上具有優勢, 而在處理不同的數據類型是具有很大的靈活性. 你甚至可以通過socket發送序列化了的CLR對象，盡管我在這里不會那樣做. 這個項目將會想你展示如何在socket之間發送文本.

代碼的運用

使用下面的代碼，你初始化了一個Server類，并運行了Start()方法:

1. Server myServer = new Server(); 
2. myServer.Start(); 

如果你計劃在一個Windows表單中管理服務器的話，我建議使用一個BackgroundWorker, 因為socket方法(一般會是ManualResentEvent) 將會阻塞GUI線程的運行.

Server 類:

1. using System.Net.Sockets; 
2.  
3. public class Server 
4. { 
5. private static Socket listener; 
6. public static ManualResetEvent allDone = new ManualResetEvent(false); 
7. public const int _bufferSize = 1024; 
8. public const int _port = 50000; 
9. public static bool _isRunning = true; 
10.  
11. class StateObject 
12. { 
13. public Socket workSocket = null; 
14. public byte[] buffer = new byte[bufferSize]; 
15. public StringBuilder sb = new StringBuilder(); 
16. } 
17.  
18. // Returns the string between str1 and str2 
19. static string Between(string str, string str1, string str2) 
20. { 
21. int i1 = 0, i2 = 0; 
22. string rtn = ""; 
23.  
24. i1 = str.IndexOf(str1, StringComparison.InvariantCultureIgnoreCase); 
25. if (i1 > -1) 
26. { 
27. i2 = str.IndexOf(str2, i1 + 1, StringComparison.InvariantCultureIgnoreCase); 
28. if (i2 > -1) 
29. { 
30. rtn = str.Substring(i1 + str1.Length, i2 - i1 - str1.Length); 
31. } 
32. } 
33. return rtn; 
34. } 
35.  
36. // Checks if the socket is connected 
37. static bool IsSocketConnected(Socket s) 
38. { 
39. return !((s.Poll(1000, SelectMode.SelectRead) && (s.Available == 0)) || !s.Connected); 
40. } 
41.  
42. // Insert all the other methods here. 
43. } 

ManualResetEvent 是一個實現了你的socket服務器中事件的.NET類. 我們需要這個項目在我們想要發布阻塞操作的時候向代碼發送信號. 你可以試驗一下用bufferSize來適配你的需求. 如果能預期到消息的大小, 使用byte單位來設置消息的大小參數bufferSize. port是偵聽TCP的端口參數. 要意識到為其它應用程序伺服所使用的接口. 如果你想要能夠方便地停止服務器，你需要實現一些機制來將_isRunning設置成false. 這一般可以借助于使用一個 BackgroundWorker做到, 其中你可以使用myWorker.CancellationPending替換_isRunning. 我提到_isRunning的原因是給你在處理取消操作的問題上提供一個方向, 并向你展示偵聽器可以方便的停止的.

Between() 和IsSocketConnected() 是輔助方法.

現在轉過來看看方法. 首先是Start()方法:

1. public void Start() 
2. { 
3. IPHostEntry ipHostInfo = Dns.GetHostEntry(Dns.GetHostName()); 
4. IPEndPoint localEP = new IPEndPoint(IPAddress.Any, _port); 
5. listener = new Socket(localEP.Address.AddressFamily, SocketType.Stream, ProtocolType.Tcp); 
6. listener.Bind(localEP); 
7.  
8. while (_IsRunning) 
9. { 
10. allDone.Reset(); 
11. listener.Listen(10); 
12. listener.BeginAccept(new AsyncCallback(acceptCallback), listener); 
13. bool isRequest = allDone.WaitOne(new TimeSpan(12, 0, 0)); // Blocks for 12 hours 
14.  
15. if (!isRequest) 
16. { 
17. allDone.Set(); 
18. // Do some work here every 12 hours 
19. } 
20. } 
21. listener.Close(); 
22. } 

這個方法初始化了偵聽器socket, 并開始等待用戶連接的到來. 項目中主要的模式是使用異步委派. 異步委派是在調用者中的狀態改變時被異步調用的方法. isRequest 告訴你WaitOne 是否已經因為有客戶端連接或者超時而退出.

如果你有大量的客戶端連接同時發生, 考慮提高Listen()方法的隊列參數.

現在來看看下一個方法, acceptCallback . 這個方法由listener.BeginAccept異步調用. 當方法完成執行時，偵聽器會立即偵聽新的客戶端.

1. static void acceptCallback(IAsyncResult ar) 
2. { 
3. // Get the listener that handles the client request. 
4. Socket listener = (Socket)ar.AsyncState; 
5.  
6. if (listener != null) 
7. { 
8. Socket handler = listener.EndAccept(ar); 
9.  
10. // Signal main thread to continue 
11. allDone.Set(); 
12.  
13. // Create state 
14. StateObject state = new StateObject(); 
15. state.workSocket = handler; 
16. handler.BeginReceive(state.buffer, 0, _bufferSize, 0, new AsyncCallback(readCallback), state); 
17. } 
18. } 

acceptCallback 會派生出另外一個異步指派: readCallback. 這個方法會讀取來自socket的實際數據. 我已經為收發數據作了我自己的控制, 對于_bufferSize來說是不變的. 所有發送到服務器的字符串都必須用和 包起來. 同樣，客戶端在收到服務器的響應式，必須解除響應信息的包裹, 后者被和 包了起來。

1. static void readCallback(IAsyncResult ar) 
2. { 
3. StateObject state = (StateObject)ar.AsyncState; 
4. Socket handler = state.workSocket; 
5.  
6. if (!IsSocketConnected(handler))  
7. { 
8. handler.Close(); 
9. return; 
10. } 
11.  
12. int read = handler.EndReceive(ar); 
13.  
14. // Data was read from the client socket. 
15. if (read > 0) 
16. { 
17. state.sb.Append(Encoding.UTF8.GetString(state.buffer, 0, read)); 
18.  
19. if (state.sb.ToString().Contains("<!--ENDSOCKET-->")) 
20. { 
21. string toSend = ""; 
22. string cmd = ts.Strings.Between(state.sb.ToString(), "<!--SOCKET-->", "<!--ENDSOCKET-->"); 
23.  
24. switch (cmd) 
25. { 
26. case "Hi!": 
27. toSend = "How are you?"; 
28. break; 
29. case "Milky Way?": 
30. toSend = "No I am not."; 
31. break; 
32. } 
33.  
34. toSend = "<!--RESPONSE-->" + toSend + "<!--ENDRESPONSE-->"; 
35.  
36. byte[] bytesToSend = Encoding.UTF8.GetBytes(toSend); 
37. handler.BeginSend(bytesToSend, 0, bytesToSend.Length, SocketFlags.None 
38. , new AsyncCallback(sendCallback), state); 
39. } 
40. else 
41. { 
42. handler.BeginReceive(state.buffer, 0, _bufferSize, 0 
43. , new AsyncCallback(readCallback), state); 
44. } 
45. } 
46. else 
47. { 
48. handler.Close(); 
49. } 
50. } 

readCallback 會派生另外一個方法, sendCallback, 它將會向客戶端發送請求. 如果客戶端沒有關閉連接, sendCallback 將會向socket發送信號以獲得更多的數據.

1. static void sendCallback(IAsyncResult ar) 
2. { 
3. StateObject state = (StateObject)ar.AsyncState; 
4. Socket handler = state.workSocket; 
5. handler.EndSend(ar); 
6.  
7. StateObject newstate = new StateObject(); 
8. newstate.workSocket = handler; 
9. handler.BeginReceive(newstate.buffer, 0, StateObject.BufferSize, 0, new AsyncCallback(readCallback), newstate); 
10. } 

我會將寫一個socket客戶端作為聯系留給讀者. socket客戶端應該使用同異步調用同樣的編程模式. 我希望你能從這篇文章中收獲樂趣，并且會像一個socket程序員那樣付諸實踐!

要點

我在生產環境下使用了此代碼，其中的socket服務器是一個自由文本搜索引擎。 SQL Server缺乏對自由文本搜索支持(你可以使用自由文本索引，但它們是緩慢和昂貴的)。socket服務器負載了大量導向IEnumerables的文本數據，并使用Linq來搜索文本。來自socket服務器的響應從數百萬行的Unicode文本數據中搜索時間在幾毫秒內。我們還使用了三個分布式的Sphinx服務器(www.sphinxsearch.com)。socket服務器充當了Sphinx服務器的高速緩存。如果你需要一個快速的自由文本搜索引擎，我強烈建議使用Sphinx。