首页 > 代码库 > Python异步Socket编程

Python异步Socket编程

异步网络据说能极大的提高网络server的连接速度,所以打算写一个专题,来学习和了解异步网络.因为Python有个非常出名的异步Lib:Twisted,所以就用Python来完成. 

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

Java代码  收藏代码
  1. import optparse  
  2. import os  
  3. import socket  
  4. import time  
  5. from threading import Thread  
  6. import StringIO  
  7.    
  8. txt = ‘‘‘1111  
  9. 2222  
  10. 3333  
  11. 4444  
  12. ‘‘‘  
  13.    
  14. def server(listen_socket):  
  15.     while True:  
  16.         buf = StringIO.StringIO(txt)  
  17.         sock, addr = listen_socket.accept()  
  18.         print ‘Somebody at %s wants poetry!‘ % (addr,)  
  19.         while True:  
  20.                 try:  
  21.                     line = buf.readline().strip()  
  22.                     if not line:  
  23.                         sock.close()  
  24.                         break  
  25.                     sock.sendall(line)  # this is a blocking call  
  26.                     print ‘send bytes to client:%s‘ % line  
  27.                     #sock.close()  
  28.                 except socket.error:  
  29.                     sock.close()  
  30.                     break  
  31.                 time.sleep(1)  #server和client连接后,server会故意每发送一个单词后等待一秒钟后再发送另一个单词  
  32.    
  33.    
  34. def main():  
  35.     ports = [10000, 10001, 10002]  
  36.     for port in ports:  
  37.         listen_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  
  38.         listen_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)  
  39.         addres = (str(‘127.0.0.1‘), port)  
  40.         listen_socket.bind(addres)  
  41.         listen_socket.listen(5)  
  42.         print "start listen at:%s" % (port,)  
  43.         worker = Thread(target = server, args = [listen_socket])  
  44.         worker.setDaemon(True)  
  45.         worker.start()  
  46.    
  47.    
  48. if __name__ == ‘__main__‘:  
  49.     main()  
  50.     while True:  
  51.         time.sleep(0.1) #如果不sleep的话,CPU会被Python完全占用了  
  52.         pass  



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

Java代码  收藏代码
  1. import socket  
  2.    
  3.    
  4. if __name__ == ‘__main__‘:  
  5.     ports = [10000, 10001, 10002]  
  6.     for port in ports:  
  7.         address = (str(‘127.0.0.1‘), port)  
  8.         sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  
  9.         sock.connect(address)  
  10.         poem = ‘‘  
  11.         while True:  
  12.             data = sock.recv(4)  
  13.             if not data:  
  14.                 sock.close()  
  15.                 break  
  16.             poem += data  
  17.         print poem  



下面用异步的client来读取,代码如下: 

Java代码  收藏代码
  1. import datetime, errno, optparse, select, socket  
  2.    
  3. def connect(port):  
  4.     """Connect to the given server and return a non-blocking socket."""  
  5.     address = (str(‘127.0.0.1‘), port)  
  6.     sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  
  7.     sock.connect(address)  
  8.     sock.setblocking(0)  
  9.     return sock  
  10.    
  11. def format_address(address):  
  12.     host, port = address  
  13.     return ‘%s:%s‘ % (host or ‘127.0.0.1‘, port)  
  14.    
  15. if __name__ == ‘__main__‘:  
  16.     ports = [10000, 10001, 10002]  
  17.     start = datetime.datetime.now()  
  18.    
  19.     sockets = map(connect, ports)  
  20.     poems = dict.fromkeys(sockets, ‘‘) # socket -> accumulated poem   
  21.    
  22.     # socket -> task numbers  
  23.     sock2task = dict([(s, i + 1) for i, s in enumerate(sockets)])  
  24.     sockets = list(sockets) # make a copy  
  25.    
  26.     while sockets:  
  27.         #运用select来确保那些可读取的异步socket可以立即开始读取IO  
  28.         #OS不停的搜索目前可以read的socket,有的话就返回rlist  
  29.         rlist, _, _ = select.select(sockets, [], [])  
  30.         for sock in rlist:  
  31.             data = ‘‘  
  32.             while True:  
  33.                 try:  
  34.                     new_data = sock.recv(1024)  
  35.                 except socket.error, e:  
  36.                     if e.args[0] == errno.EWOULDBLOCK:  
  37.                         break  
  38.                     raise  
  39.                 else:  
  40.                     if not new_data:  
  41.                         break  
  42.                     else:  
  43.                         print new_data  
  44.                         data += new_data  
  45.    
  46.             task_num = sock2task[sock]  
  47.             if not data:  
  48.                 sockets.remove(sock)  
  49.                 sock.close()  
  50.                 print ‘Task %d finished‘ % task_num  
  51.             else:  
  52.                 addr_fmt = format_address(sock.getpeername())  
  53.                 msg = ‘Task %d: got %d bytes of poetry from %s‘  
  54.                 print  msg % (task_num, len(data), addr_fmt)  
  55.    
  56.             poems[sock] += data  
  57.    
  58.     elapsed = datetime.datetime.now() - start  
  59.     print ‘Got poems in %s‘ %  elapsed  



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

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