Python异步网络探

import optparse
import os
import socket
import time
from threading import Thread
import StringIO
 
txt = &＃39;&＃39;&＃39;1111
2222
3333
4444
&＃39;&＃39;&＃39;
 
def server(listen_socket):
    while True:
        buf = StringIO.StringIO(txt)
        sock, addr = listen_socket.accept()
        print &＃39;Somebody at %s wants poetry!&＃39; % (addr,)
        while True:
                try:
                    line = buf.readline().strip()
                    if not line:
                        sock.close()
                        break
                    sock.sendall(line)  # this is a blocking call
                    print &＃39;send bytes to client:%s&＃39; % 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(&＃39;127.0.0.1&＃39;), 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__ == &＃39;__main__&＃39;:
    main()
    while True:
        time.sleep(0.1) #如果不sleep的话,CPU会被Python完全占用了
        pass

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

import socket
 
 
if __name__ == &＃39;__main__&＃39;:
    ports = [10000, 10001, 10002]
    for port in ports:
        address = (str(&＃39;127.0.0.1&＃39;), port)
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.connect(address)
        poem = &＃39;&＃39;
        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(&＃39;127.0.0.1&＃39;), 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 &＃39;%s:%s&＃39; % (host or &＃39;127.0.0.1&＃39;, port)
 
if __name__ == &＃39;__main__&＃39;:
    ports = [10000, 10001, 10002]
    start = datetime.datetime.now()
 
    sockets = map(connect, ports)
    poems = dict.fromkeys(sockets, &＃39;&＃39;) # 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 = &＃39;&＃39;
            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 &＃39;Task %d finished&＃39; % task_num
            else:
                addr_fmt = format_address(sock.getpeername())
                msg = &＃39;Task %d: got %d bytes of poetry from %s&＃39;
                print  msg % (task_num, len(data), addr_fmt)
 
            poems[sock] += data
 
    elapsed = datetime.datetime.now() - start
    print &＃39;Got poems in %s&＃39; %  elapsed

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

同步模式下，客户端分别创建socket；而在异步模式下，client开始就创建了所有的socket。

通过“sock.setblocking(0)”设置socket为异步模式。

通过Unix系统的select俩返回可读取IO

最为核心的是26行和29行。尤其是29行的select操作返回待读取socket的列表。