在Python中使用異步Socket編程性能測試

異步網絡據說能極大的提高網絡server的連接速度,所以打算寫一個專題,來學習和了解異步網絡.因為Python有個非常出名的異步Lib:Twisted,所以就用Python來完成.
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的列表。