在Python中使用异步Socket编程性能测试

OK,首先写一个python socket的server段,对开放三个端口:10000,10001,10002.krondo的例子中是每个server绑定一个端口,测试的时候需要分别开3个shell,分别运行.这太麻烦了,就分别用三个Thread来运行这些services.

    import optparse 
    import os 
    import socket 
    import time 
    from threading import Thread 
    import StringIO 

    txt = '''1111 
    2222 
    3333 
    4444 
    ''' 

      def server(listen_socket): 
      while True: 
        buf = StringIO.StringIO(txt) 
        sock, addr = listen_socket.accept() 
        print 'Somebody at %s wants poetry!' % (addr,) 
        while True: 
            try: 
              line = buf.readline().strip() 
              if not line: 
                sock.close() 
                break 
              sock.sendall(line) # this is a blocking call 
              print 'send bytes to client:%s' % line 
              #sock.close() 
            except socket.error: 
              sock.close() 
              break 
            time.sleep(1) #server和client连接后，server会故意每发送一个单词后等待一秒钟后再发送另一个单词 

      def main(): 
      ports = [10000, 10001, 10002] 
      for port in ports: 
        listen_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 
        listen_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) 
        addres = (str('127.0.0.1'), port) 
        listen_socket.bind(addres) 
        listen_socket.listen(5) 
        print "start listen at:%s" % (port,) 
        worker = Thread(target = server, args = [listen_socket]) 
        worker.setDaemon(True) 
        worker.start() 

      if __name__ == '__main__': 
      main() 
      while True: 
        time.sleep(0.1) #如果不sleep的话,CPU会被Python完全占用了 
        pass 

下面是一个client,没有才用异步网络,连接这个三个端口的server:

    import socket 


    if __name__ == '__main__': 
      ports = [10000, 10001, 10002] 
      for port in ports: 
        address = (str('127.0.0.1'), port) 
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 
        sock.connect(address) 
        poem = '' 
        while True: 
          data = sock.recv(4) 
          if not data: 
            sock.close() 
            break 
          poem += data 
        print poem 

下面用异步的client来读取，代码如下：

    import datetime, errno, optparse, select, socket 

    def connect(port): 
      """Connect to the given server and return a non-blocking socket.""" 
      address = (str('127.0.0.1'), port) 
      sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 
      sock.connect(address) 
      sock.setblocking(0) 
      return sock 

    def format_address(address): 
      host, port = address 
      return '%s:%s' % (host or '127.0.0.1', port) 

    if __name__ == '__main__': 
      ports = [10000, 10001, 10002] 
      start = datetime.datetime.now() 

      sockets = map(connect, ports) 
      poems = dict.fromkeys(sockets, '') # socket -> accumulated poem  

      # socket -> task numbers 
      sock2task = dict([(s, i + 1) for i, s in enumerate(sockets)]) 
      sockets = list(sockets) # make a copy 

      while sockets: 
        #运用select来确保那些可读取的异步socket可以立即开始读取IO 
        #OS不停的搜索目前可以read的socket，有的话就返回rlist 
        rlist, _, _ = select.select(sockets, [], []) 
        for sock in rlist: 
          data = '' 
          while True: 
            try: 
              new_data = sock.recv(1024) 
            except socket.error, e: 
              if e.args[0] == errno.EWOULDBLOCK: 
                break 
              raise 
            else: 
              if not new_data: 
                break 
              else: 
                print new_data 
                data += new_data 

          task_num = sock2task[sock] 
          if not data: 
            sockets.remove(sock) 
            sock.close() 
            print 'Task %d finished' % task_num 
          else: 
            addr_fmt = format_address(sock.getpeername()) 
            msg = 'Task %d: got %d bytes of poetry from %s' 
            print msg % (task_num, len(data), addr_fmt) 

          poems[sock] += data 

      elapsed = datetime.datetime.now() - start 
      print 'Got poems in %s' % elapsed 

结果只需要4秒就完成了读取任务。效率是刚才同步socket的三倍。对客户端的异步改造主要有两点：

同步模式下，客户端分别创建socket；而在异步模式下，client开始就创建了所有的socket。
通过"sock.setblocking(0)"设置socket为异步模式。
通过Unix系统的select俩返回可读取IO
最为核心的是26行和29行。尤其是29行的select操作返回待读取socket的列表。