使用C#来编写一个异步的Socket服务器
介绍
我最近需要为一个.net项目准备一个内部线程通信机制.项目有多个使用ASP.NET,Windows表单和控制台应用程序的服务器和客户端构成.考虑到实现的可能性,我下定决心要使用原生的socket,而不是许多.NET中已经提前为我们构建好的组件,像是所谓的管道,NetTcpClient还有Azure服务总线.
这篇文章中的服务器基于System.Net.Sockets类异步方法.这些允许你支持大量的socket客户端,而一个客户端的连接是唯一的阻塞机制.阻塞的时间是可以忽略不记得,所以服务器基本上是在当做一个多线程socket服务器在运作的.
背景
原生的socket在为你提供通信层面的完全控制权上具有优势,而在处理不同的数据类型是具有很大的灵活性.你甚至可以通过socket发送序列化了的CLR对象,尽管我在这里不会那样做.这个项目将会想你展示如何在socket之间发送文本.
代码的运用
使用下面的代码,你初始化了一个Server类,并运行了Start()方法:
ServermyServer=newServer(); myServer.Start();
如果你计划在一个Windows表单中管理服务器的话,我建议使用一个BackgroundWorker,因为socket方法(一般会是ManualResentEvent)将会阻塞GUI线程的运行.
Server类:
usingSystem.Net.Sockets; publicclassServer { privatestaticSocketlistener; publicstaticManualResetEventallDone=newManualResetEvent(false); publicconstint_bufferSize=1024; publicconstint_port=50000; publicstaticbool_isRunning=true; classStateObject { publicSocketworkSocket=null; publicbyte[]buffer=newbyte[bufferSize]; publicStringBuildersb=newStringBuilder(); } //Returnsthestringbetweenstr1andstr2 staticstringBetween(stringstr,stringstr1,stringstr2) { inti1=0,i2=0; stringrtn=""; 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); } } returnrtn; } //Checksifthesocketisconnected staticboolIsSocketConnected(Sockets) { return!((s.Poll(1000,SelectMode.SelectRead)&&(s.Available==0))||!s.Connected); } //Insertalltheothermethodshere. }
ManualResetEvent是一个实现了你的socket服务器中事件的.NET类.我们需要这个项目在我们想要发布阻塞操作的时候向代码发送信号.你可以试验一下用bufferSize来适配你的需求.如果能预期到消息的大小,使用byte单位来设置消息的大小参数bufferSize.port是侦听TCP的端口参数.要意识到为其它应用程序伺服所使用的接口.如果你想要能够方便地停止服务器,你需要实现一些机制来将_isRunning设置成false.这一般可以借助于使用一个BackgroundWorker做到,其中你可以使用myWorker.CancellationPending替换_isRunning.我提到_isRunning的原因是给你在处理取消操作的问题上提供一个方向,并向你展示侦听器可以方便的停止的.
Between()和IsSocketConnected()是辅助方法.
现在转过来看看方法.首先是Start()方法:
publicvoidStart() { IPHostEntryipHostInfo=Dns.GetHostEntry(Dns.GetHostName()); IPEndPointlocalEP=newIPEndPoint(IPAddress.Any,_port); listener=newSocket(localEP.Address.AddressFamily,SocketType.Stream,ProtocolType.Tcp); listener.Bind(localEP); while(_IsRunning) { allDone.Reset(); listener.Listen(10); listener.BeginAccept(newAsyncCallback(acceptCallback),listener); boolisRequest=allDone.WaitOne(newTimeSpan(12,0,0));//Blocksfor12hours if(!isRequest) { allDone.Set(); //Dosomeworkhereevery12hours } } listener.Close(); }
这个方法初始化了侦听器socket,并开始等待用户连接的到来.项目中主要的模式是使用异步委派.异步委派是在调用者中的状态改变时被异步调用的方法.isRequest告诉你WaitOne是否已经因为有客户端连接或者超时而退出.
如果你有大量的客户端连接同时发生,考虑提高Listen()方法的队列参数.
现在来看看下一个方法,acceptCallback.这个方法由listener.BeginAccept异步调用.当方法完成执行时,侦听器会立即侦听新的客户端.
staticvoidacceptCallback(IAsyncResultar) { //Getthelistenerthathandlestheclientrequest. Socketlistener=(Socket)ar.AsyncState; if(listener!=null) { Sockethandler=listener.EndAccept(ar); //Signalmainthreadtocontinue allDone.Set(); //Createstate StateObjectstate=newStateObject(); state.workSocket=handler; handler.BeginReceive(state.buffer,0,_bufferSize,0,newAsyncCallback(readCallback),state); } }
acceptCallback会派生出另外一个异步指派:readCallback.这个方法会读取来自socket的实际数据.我已经为收发数据作了我自己的控制,对于_bufferSize来说是不变的.所有发送到服务器的字符串都必须用<!--SOCKET-->和<!--ENDSOCKET-->包起来.同样,客户端在收到服务器的响应式,必须解除响应信息的包裹,后者被<!--RESPONSE-->和<!--ENDRESPONSE-->包了起来。
staticvoidreadCallback(IAsyncResultar) { StateObjectstate=(StateObject)ar.AsyncState; Sockethandler=state.workSocket; if(!IsSocketConnected(handler)) { handler.Close(); return; } intread=handler.EndReceive(ar); //Datawasreadfromtheclientsocket. if(read>0) { state.sb.Append(Encoding.UTF8.GetString(state.buffer,0,read)); if(state.sb.ToString().Contains("<!--ENDSOCKET-->")) { stringtoSend=""; stringcmd=ts.Strings.Between(state.sb.ToString(),"<!--SOCKET-->","<!--ENDSOCKET-->"); switch(cmd) { case"Hi!": toSend="Howareyou?"; break; case"MilkyWay?": toSend="NoIamnot."; break; } toSend="<!--RESPONSE-->"+toSend+"<!--ENDRESPONSE-->"; byte[]bytesToSend=Encoding.UTF8.GetBytes(toSend); handler.BeginSend(bytesToSend,0,bytesToSend.Length,SocketFlags.None ,newAsyncCallback(sendCallback),state); } else { handler.BeginReceive(state.buffer,0,_bufferSize,0 ,newAsyncCallback(readCallback),state); } } else { handler.Close(); } }
readCallback会派生另外一个方法,sendCallback,它将会向客户端发送请求.如果客户端没有关闭连接,sendCallback将会向socket发送信号以获得更多的数据.
staticvoidsendCallback(IAsyncResultar) { StateObjectstate=(StateObject)ar.AsyncState; Sockethandler=state.workSocket; handler.EndSend(ar); StateObjectnewstate=newStateObject(); newstate.workSocket=handler; handler.BeginReceive(newstate.buffer,0,StateObject.BufferSize,0,newAsyncCallback(readCallback),newstate); }
我会将写一个socket客户端作为联系留给读者.socket客户端应该使用同异步调用同样的编程模式.我希望你能从这篇文章中收获乐趣,并且会像一个socket程序员那样付诸实践!
要点
我在生产环境下使用了此代码,其中的socket服务器是一个自由文本搜索引擎。SQLServer缺乏对自由文本搜索支持(你可以使用自由文本索引,但它们是缓慢和昂贵的)。socket服务器负载了大量导向IEnumerables的文本数据,并使用Linq来搜索文本。来自socket服务器的响应从数百万行的Unicode文本数据中搜索时间在几毫秒内。我们还使用了三个分布式的Sphinx服务器(www.sphinxsearch.com)。socket服务器充当了Sphinx服务器的高速缓存。如果你需要一个快速的自由文本搜索引擎,我强烈建议使用Sphinx。