欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

使用C#来编写一个异步的Socket服务器

程序员文章站 2023-11-20 22:11:16
介绍 我最近需要为一个.net项目准备一个内部线程通信机制. 项目有多个使用asp.net,windows 表单和控制台应用程序的服务器和客户端构成. 考虑到实现的可能性...

介绍

我最近需要为一个.net项目准备一个内部线程通信机制. 项目有多个使用asp.net,windows 表单和控制台应用程序的服务器和客户端构成. 考虑到实现的可能性,我下定决心要使用原生的socket,而不是许多.net中已经提前为我们构建好的组件, 像是所谓的管道, nettcpclient 还有 azure 服务总线.

这篇文章中的服务器基于system.net.sockets类异步方法. 这些允许你支持大量的socket客户端, 而一个客户端的连接是唯一的阻塞机制. 阻塞的时间是可以忽略不记得,所以服务器基本上是在当做一个多线程socket服务器在运作的.

背景

原生的socket在为你提供通信层面的完全控制权上具有优势, 而在处理不同的数据类型是具有很大的灵活性. 你甚至可以通过socket发送序列化了的clr对象,尽管我在这里不会那样做. 这个项目将会想你展示如何在socket之间发送文本.
代码的运用

使用下面的代码,你初始化了一个server类,并运行了start()方法:
 

server myserver = new server();
myserver.start();

如果你计划在一个windows表单中管理服务器的话,我建议使用一个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 是一个实现了你的socket服务器中事件的.net类. 我们需要这个项目在我们想要发布阻塞操作的时候向代码发送信号. 你可以试验一下用buffersize来适配你的需求. 如果能预期到消息的大小, 使用byte单位来设置消息的大小参数buffersize. port是侦听tcp的端口参数. 要意识到为其它应用程序伺服所使用的接口. 如果你想要能够方便地停止服务器,你需要实现一些机制来将_isrunning设置成false. 这一般可以借助于使用一个 backgroundworker做到, 其中你可以使用myworker.cancellationpending替换_isrunning. 我提到_isrunning的原因是给你在处理取消操作的问题上提供一个方向, 并向你展示侦听器可以方便的停止的.

between() 和issocketconnected() 是辅助方法.


现在转过来看看方法. 首先是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, 并开始等待用户连接的到来. 项目中主要的模式是使用异步委派. 异步委派是在调用者中的状态改变时被异步调用的方法. isrequest 告诉你waitone 是否已经因为有客户端连接或者超时而退出.

如果你有大量的客户端连接同时发生, 考虑提高listen()方法的队列参数.


现在来看看下一个方法, acceptcallback . 这个方法由listener.beginaccept异步调用. 当方法完成执行时,侦听器会立即侦听新的客户端.
 

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的实际数据. 我已经为收发数据作了我自己的控制, 对于_buffersize来说是不变的. 所有发送到服务器的字符串都必须用<!--socket--> 和 <!--endsocket-->包起来. 同样,客户端在收到服务器的响应式,必须解除响应信息的包裹, 后者被<!--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, 它将会向客户端发送请求. 如果客户端没有关闭连接, sendcallback 将会向socket发送信号以获得更多的数据.
 

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客户端作为联系留给读者. socket客户端应该使用同异步调用同样的编程模式. 我希望你能从这篇文章中收获乐趣,并且会像一个socket程序员那样付诸实践!

要点

我在生产环境下使用了此代码,其中的socket服务器是一个*文本搜索引擎。 sql server缺乏对*文本搜索支持(你可以使用*文本索引,但它们是缓慢和昂贵的)。socket服务器负载了大量导向ienumerables的文本数据,并使用linq来搜索文本。来自socket服务器的响应从数百万行的unicode文本数据中搜索时间在几毫秒内。我们还使用了三个分布式的sphinx服务器(www.sphinxsearch.com)。socket服务器充当了sphinx服务器的高速缓存。如果你需要一个快速的*文本搜索引擎,我强烈建议使用sphinx。