使用C#來編寫一個異步的Socket服務(wù)器
介紹
我最近需要為一個.net項目準(zhǔn)備一個內(nèi)部線程通信機(jī)制. 項目有多個使用ASP.NET,Windows 表單和控制臺應(yīng)用程序的服務(wù)器和客戶端構(gòu)成. 考慮到實現(xiàn)的可能性,我下定決心要使用原生的socket,而不是許多.NET中已經(jīng)提前為我們構(gòu)建好的組件, 像是所謂的管道, NetTcpClient 還有 Azure 服務(wù)總線.
這篇文章中的服務(wù)器基于System.Net.Sockets類異步方法. 這些允許你支持大量的socket客戶端, 而一個客戶端的連接是唯一的阻塞機(jī)制. 阻塞的時間是可以忽略不記得,所以服務(wù)器基本上是在當(dāng)做一個多線程socket服務(wù)器在運作的.
背景
原生的socket在為你提供通信層面的完全控制權(quán)上具有優(yōu)勢, 而在處理不同的數(shù)據(jù)類型是具有很大的靈活性. 你甚至可以通過socket發(fā)送序列化了的CLR對象,盡管我在這里不會那樣做. 這個項目將會想你展示如何在socket之間發(fā)送文本.
代碼的運用
使用下面的代碼,你初始化了一個Server類,并運行了Start()方法:
Server myServer = new Server(); myServer.Start();
如果你計劃在一個Windows表單中管理服務(wù)器的話,我建議使用一個BackgroundWorker, 因為socket方法(一般會是ManualResentEvent) 將會阻塞GUI線程的運行.
Server 類:
using System.Net.Sockets;
public class Server
{
private static Socket listener;
public static ManualResetEvent allDone = new ManualResetEvent(false);
public const int _bufferSize = 1024;
public const int _port = 50000;
public static bool _isRunning = true;
class StateObject
{
public Socket workSocket = null;
public byte[] buffer = new byte[bufferSize];
public StringBuilder sb = new StringBuilder();
}
// Returns the string between str1 and str2
static string Between(string str, string str1, string str2)
{
int i1 = 0, i2 = 0;
string rtn = "";
i1 = str.IndexOf(str1, StringComparison.InvariantCultureIgnoreCase);
if (i1 > -1)
{
i2 = str.IndexOf(str2, i1 + 1, StringComparison.InvariantCultureIgnoreCase);
if (i2 > -1)
{
rtn = str.Substring(i1 + str1.Length, i2 - i1 - str1.Length);
}
}
return rtn;
}
// Checks if the socket is connected
static bool IsSocketConnected(Socket s)
{
return !((s.Poll(1000, SelectMode.SelectRead) && (s.Available == 0)) || !s.Connected);
}
// Insert all the other methods here.
}
ManualResetEvent 是一個實現(xiàn)了你的socket服務(wù)器中事件的.NET類. 我們需要這個項目在我們想要發(fā)布阻塞操作的時候向代碼發(fā)送信號. 你可以試驗一下用bufferSize來適配你的需求. 如果能預(yù)期到消息的大小, 使用byte單位來設(shè)置消息的大小參數(shù)bufferSize. port是偵聽TCP的端口參數(shù). 要意識到為其它應(yīng)用程序伺服所使用的接口. 如果你想要能夠方便地停止服務(wù)器,你需要實現(xiàn)一些機(jī)制來將_isRunning設(shè)置成false. 這一般可以借助于使用一個 BackgroundWorker做到, 其中你可以使用myWorker.CancellationPending替換_isRunning. 我提到_isRunning的原因是給你在處理取消操作的問題上提供一個方向, 并向你展示偵聽器可以方便的停止的.
Between() 和IsSocketConnected() 是輔助方法.
現(xiàn)在轉(zhuǎn)過來看看方法. 首先是Start()方法:
public void Start()
{
IPHostEntry ipHostInfo = Dns.GetHostEntry(Dns.GetHostName());
IPEndPoint localEP = new IPEndPoint(IPAddress.Any, _port);
listener = new Socket(localEP.Address.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
listener.Bind(localEP);
while (_IsRunning)
{
allDone.Reset();
listener.Listen(10);
listener.BeginAccept(new AsyncCallback(acceptCallback), listener);
bool isRequest = allDone.WaitOne(new TimeSpan(12, 0, 0)); // Blocks for 12 hours
if (!isRequest)
{
allDone.Set();
// Do some work here every 12 hours
}
}
listener.Close();
}
這個方法初始化了偵聽器socket, 并開始等待用戶連接的到來. 項目中主要的模式是使用異步委派. 異步委派是在調(diào)用者中的狀態(tài)改變時被異步調(diào)用的方法. isRequest 告訴你WaitOne 是否已經(jīng)因為有客戶端連接或者超時而退出.
如果你有大量的客戶端連接同時發(fā)生, 考慮提高Listen()方法的隊列參數(shù).
現(xiàn)在來看看下一個方法, acceptCallback . 這個方法由listener.BeginAccept異步調(diào)用. 當(dāng)方法完成執(zhí)行時,偵聽器會立即偵聽新的客戶端.
static void acceptCallback(IAsyncResult ar)
{
// Get the listener that handles the client request.
Socket listener = (Socket)ar.AsyncState;
if (listener != null)
{
Socket handler = listener.EndAccept(ar);
// Signal main thread to continue
allDone.Set();
// Create state
StateObject state = new StateObject();
state.workSocket = handler;
handler.BeginReceive(state.buffer, 0, _bufferSize, 0, new AsyncCallback(readCallback), state);
}
}
acceptCallback 會派生出另外一個異步指派: readCallback. 這個方法會讀取來自socket的實際數(shù)據(jù). 我已經(jīng)為收發(fā)數(shù)據(jù)作了我自己的控制, 對于_bufferSize來說是不變的. 所有發(fā)送到服務(wù)器的字符串都必須用<!--SOCKET--> 和 <!--ENDSOCKET-->包起來. 同樣,客戶端在收到服務(wù)器的響應(yīng)式,必須解除響應(yīng)信息的包裹, 后者被<!--RESPONSE--> 和 <!--ENDRESPONSE-->包了起來。
static void readCallback(IAsyncResult ar)
{
StateObject state = (StateObject)ar.AsyncState;
Socket handler = state.workSocket;
if (!IsSocketConnected(handler))
{
handler.Close();
return;
}
int read = handler.EndReceive(ar);
// Data was read from the client socket.
if (read > 0)
{
state.sb.Append(Encoding.UTF8.GetString(state.buffer, 0, read));
if (state.sb.ToString().Contains("<!--ENDSOCKET-->"))
{
string toSend = "";
string cmd = ts.Strings.Between(state.sb.ToString(), "<!--SOCKET-->", "<!--ENDSOCKET-->");
switch (cmd)
{
case "Hi!":
toSend = "How are you?";
break;
case "Milky Way?":
toSend = "No I am not.";
break;
}
toSend = "<!--RESPONSE-->" + toSend + "<!--ENDRESPONSE-->";
byte[] bytesToSend = Encoding.UTF8.GetBytes(toSend);
handler.BeginSend(bytesToSend, 0, bytesToSend.Length, SocketFlags.None
, new AsyncCallback(sendCallback), state);
}
else
{
handler.BeginReceive(state.buffer, 0, _bufferSize, 0
, new AsyncCallback(readCallback), state);
}
}
else
{
handler.Close();
}
}
readCallback 會派生另外一個方法, sendCallback, 它將會向客戶端發(fā)送請求. 如果客戶端沒有關(guān)閉連接, sendCallback 將會向socket發(fā)送信號以獲得更多的數(shù)據(jù).
static void sendCallback(IAsyncResult ar)
{
StateObject state = (StateObject)ar.AsyncState;
Socket handler = state.workSocket;
handler.EndSend(ar);
StateObject newstate = new StateObject();
newstate.workSocket = handler;
handler.BeginReceive(newstate.buffer, 0, StateObject.BufferSize, 0, new AsyncCallback(readCallback), newstate);
}
我會將寫一個socket客戶端作為聯(lián)系留給讀者. socket客戶端應(yīng)該使用同異步調(diào)用同樣的編程模式. 我希望你能從這篇文章中收獲樂趣,并且會像一個socket程序員那樣付諸實踐!
要點
我在生產(chǎn)環(huán)境下使用了此代碼,其中的socket服務(wù)器是一個自由文本搜索引擎。 SQL Server缺乏對自由文本搜索支持(你可以使用自由文本索引,但它們是緩慢和昂貴的)。socket服務(wù)器負(fù)載了大量導(dǎo)向IEnumerables的文本數(shù)據(jù),并使用Linq來搜索文本。來自socket服務(wù)器的響應(yīng)從數(shù)百萬行的Unicode文本數(shù)據(jù)中搜索時間在幾毫秒內(nèi)。我們還使用了三個分布式的Sphinx服務(wù)器(www.sphinxsearch.com)。socket服務(wù)器充當(dāng)了Sphinx服務(wù)器的高速緩存。如果你需要一個快速的自由文本搜索引擎,我強(qiáng)烈建議使用Sphinx。
相關(guān)文章
C#?wpf實現(xiàn)任意控件更多調(diào)整大小功能
這篇文章主要為大家詳細(xì)介紹了C#?wpf實現(xiàn)任意控件更多調(diào)整大小功能的相關(guān)知識,文中的示例代碼講解詳細(xì),感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下2024-01-01

