Python IO多路复用

首先什么是I/O：

I/O（input/output），即输入/输出端口。每个设备都会有一个专用的I/O地址，用来处理自己的输入输出信息

I/O分为磁盘io和网络io，这里说的是网络io

IO多路复用:

I/O多路复用指：通过一种机制，可以监视多个描述符(socket)，一旦某个描述符就绪（一般是读就绪或者写就绪），能够通知程序进行相应的读写操作。

Linux

Linux中的 select，poll，epoll 都是IO多路复用的机制。

Linux下网络I/O使用socket套接字来通信，普通I/O模型只能监听一个socket，而I/O多路复用可同时监听多个socket.

I/O多路复用避免阻塞在io上，原本为多进程或多线程来接收多个连接的消息变为单进程或单线程保存多个socket的状态后轮询处理.

Python  

Python中有一个select模块，其中提供了：select、poll、epoll三个方法，分别调用系统的 select，poll，epoll 从而实现IO多路复用。

Windows Python：

    提供： select

Mac Python：

    提供： select

Linux Python：

    提供： select、poll、epoll

对于select模块操作的方法：

句柄列表11, 句柄列表22, 句柄列表33 = select.select(句柄序列1, 句柄序列2, 句柄序列3, 超时时间)
 
参数： 可接受四个参数（前三个必须）
返回值：三个列表
 
select方法用来监视文件句柄，如果句柄发生变化，则获取该句柄。
1、当 参数1 序列中的句柄发生可读时（accetp和read），则获取发生变化的句柄并添加到 返回值1 序列中
2、当 参数2 序列中含有句柄时，则将该序列中所有的句柄添加到 返回值2 序列中
3、当 参数3 序列中的句柄发生错误时，则将该发生错误的句柄添加到 返回值3 序列中
4、当 超时时间 未设置，则select会一直阻塞，直到监听的句柄发生变化
5、当 超时时间 ＝ 1时，那么如果监听的句柄均无任何变化，则select会阻塞 1 秒，之后返回三个空列表，如果监听的句柄有变化，则直接执行。

利用select监听终端操作实例1：

#!/usr/bin/env python
# -*- coding:utf-8 -*- 
import select
import sys

while True:
    readable, writeable, error = select.select([sys.stdin,],[],[],1)
    """
    select.select([sys.stdin,],[],[],1)就是I/O多路复用的例子，第一个参数是列表，这里放进去的是sys.stdin就是我输入
    进去东西的描述符.第一个参数对应的是readable这个句柄，后面两个先不考虑.

     """
    if sys.stdin in readable:
        print 'select get stdin',sys.stdin.readline()

#注意：[sys.stdin,]  这个参数不管是列表还是元组在最后的元素后面建议增加一个逗号，是一种写作习惯.
#这个代码可以在linux平台执行,windows不支持select方法的epoll方法.只支持select socket 方法.

利用select监听终端操作实例2:

#/usr/bin/env python
#-*- coding:utf-8 -*-

import socket
import select
#创建socket对象
sk = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
sk.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)      #setsockopt获得端口重用
#设置监听的IP与端口
sk.bind(('127.0.0.1',8000))
#设置client最大等待连接数
sk.listen(5)
sk.setblocking(False)   #这里设置setblocking为Falseaccept将不在阻塞，但是如果没有收到请求就会报错
while True:
    readable_list, writeable_list, error_list = select.select([sk,],[],[],2)  #监听第一个列表的文件描述符，如果里面有文件描述符发生改变既能捕获并放到readable_list中
    for r in readable_list:    #如果是空列表将不执行，如果是非空列表。将执行。
        conn,addr = r.accept()
        print addr

##################################################
#执行程序并打开IE输入地址：127.0.0.1:8000 输出如下：
('127.0.0.1', 52155)

利用select实现伪同时处理多个Socket客户端请求：

#/usr/bin/env python
#-*- coding:utf-8 -*-
import time
import socket
import select
#创建socket对象
sk = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
sk.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)      #setsockopt获得端口重用
#设置监听的IP与端口
sk.bind(('127.0.0.1',8000))
#设置client最大等待连接数
sk.listen(3)
sk.setblocking(False)   #这里设置setblocking为False accept将不在阻塞，但是如果没有收到请求就会报错
inputs = [sk,]          #将sk这个对象加入到列表中，并且赋值给inputs
#原因：看上例conn是客户端对象，客户是一直连接着呢，连接的时候状态变了，连接上之后，还是服务端的socket 有关吗？
#是不是的把他改为动态的？

while True:
    readable_list, writeable_list, error_list = select.select(inputs,[],[],1)  #把第一个参数设为列表动态的添加
    time.sleep(2)   #暂停2秒，测试使用
    print "inputs list :",inputs                # 打印inputs列表，查看执行变化
    print "file descriptor :",readable_list     #打印readable_list ，查看执行变化

    for r in readable_list:
        if r == sk:  #这里判断，如果是客户端连接过来的话他不是sk，如果是服务端的socket连接过来的话是sk
            conn,address = r.accept()
            inputs.append(conn)
            print address
        else:
        #如果是客户端，接受和返回数据
            client_data = r.recv(1024)
            r.sendall(client_data)

#!/usr/bin/env python
#-*- coding:utf-8 -*-

import socket

client = socket.socket()
client.connect(('127.0.0.1',8000))
client.settimeout(5)

while True:
    client_input = raw_input('please input message:').strip()
    client.sendall(client_input)
    server_data = client.recv(1024)
    print server_data

交互过程：

#1  默认，sk这个对象文件句柄就在inputs列表中select监听客户端的请求，当有客户端请求过来 client1 ---> server
#用户捕获了变化readable_list = [sk,]  那么循环是有值得，判断r = sk 说明是一个新的请求链接，然后把client链接加入到inputs里 inputs = [sk,conn1,]
#如果现在什么都不做，那么select无法捕获到变化：readable_list = []
#执行看下：
inputs list : [<socket._socketobject object at 0x0000000002C66798>] #默认inputs list 就有一个server socket sk 对象
file descriptor : [<socket._socketobject object at 0x0000000002C66798>]  #当有客户端请求过来时候，sk发生了变化，select捕获到了
('127.0.0.1', 62495)
inputs list : [<socket._socketobject object at 0x0000000002C66798>, <socket._socketobject object at 0x0000000002C66800>]  #第二次循环的时候，inputs = [sk,conn1,]
file descriptor : [] #第二次循环的时候readable_list = [] 因为客户端没有做任何操作，没有捕获到变化所以为空

#2 又有一个新的链接过来了，谁变化了？  sk 他变化了，有人向他发起了一个请求链接，那么现在inputs = [sk,conn1,conn2]  readable_list = [sk]
#本次循环完成之后再循环的时候 inputs = [sk,conn1,conn2,]  readable_list = [] 因为我们没有继续做操作

#第一个链接
inputs list : [<socket._socketobject object at 0x0000000002C56798>]  #默认只有一个对象
file descriptor : []
inputs list : [<socket._socketobject object at 0x0000000002C56798>]  
file descriptor : [<socket._socketobject object at 0x0000000002C56798>] #当捕获到，判断是否是新链接，如果是加入到inputs列表中监控
('127.0.0.1', 62539)
inputs list : [<socket._socketobject object at 0x0000000002C56798>, <socket._socketobject object at 0x0000000002C56800>]  #inputs列表变更为了[sk,conn1]
file descriptor : []  #因为没有后续的操作，这里没有捕获到异常所以列表为空

#第二个链接
inputs list : [<socket._socketobject object at 0x0000000002C56798>, <socket._socketobject object at 0x0000000002C56800>]  #第一个链接没有做任何操作
file descriptor : [<socket._socketobject object at 0x0000000002C56798>] #第二个链接过来了被捕获到，判断是否为新链接
('127.0.0.1', 62548)
inputs list : [<socket._socketobject object at 0x0000000002C56798>, <socket._socketobject object at 0x0000000002C56800>, <socket._socketobject object at 0x0000000002C56868>] #加入到inputs列表中
file descriptor : []
inputs list : [<socket._socketobject object at 0x0000000002C56798>, <socket._socketobject object at 0x0000000002C56800>, <socket._socketobject object at 0x0000000002C56868>]
file descriptor : []
inputs list : [<socket._socketobject object at 0x0000000002C56798>, <socket._socketobject object at 0x0000000002C56800>, <socket._socketobject object at 0x0000000002C56868>]
file descriptor : []

优化：当client端退出后，在inputs列表中移除对象！

#/usr/bin/env python
#-*- coding:utf-8 -*-
import time
import socket
import select
#创建socket对象
sk = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
sk.setsockopt
#设置监听的IP与端口
sk.bind(('127.0.0.1',6666))
#设置client最大等待连接数
sk.listen(5)
sk.setblocking(False) #这里设置setblocking为Falseaccept将不在阻塞，但是如果没有收到请求就会报错
inputs = [sk,] #将sk这个对象加入到列表中，并且赋值给inputs
#原因：看上例conn是客户端对象，客户是一直连接着呢，连接的时候状态变了，连接上之后，连接上之后，还是服务端的socket 有关吗？
#是不是的把他改为动态的？

while True:
    readable_list, writeable_list, error_list = select.select(inputs,[],[],1)  #把第一个参数设为列表动态的添加
    time.sleep(2) #测试使用
    print "inputs list :",inputs     #打印inputs列表，查看执行变化
    print "file descriptor :",readable_list #打印readable_list ，查看执行变化

    for r in readable_list:
        if r == sk:  #这里判断，如果是客户端连接过来的话他不是sk，如果是服务端的socket连接过来的话是sk
            conn,address = r.accept()
            inputs.append(conn)
            print address
        else:
        #如果是客户端，接受和返回数据
            client_data = r.recv(1024)
            if client_data:
                r.sendall(client_data)
            else:
                inputs.remove(r)#如果没有收到客户端端数据，则移除客户端句柄 因为，不管是正常关闭还是异常关闭，client端的系统底层都会发送一个消息

select socket server - server release client-connect

通过I/O多路复用让socket实现了处理多个客户端的方法，参数注解：

#第一个参数，监听的句柄序列，当有变动的时候就能捕获到把值赋值给readable_list
#如果第二参数有参数，即只要不是空列表，select就能感知，然后writeabled_list就能获取值
#第三个参数监听描述符，select内部，检测列表里面的描述符在底层操作的时候有没有异常，如果异常了他也当成一个变化，把这个赋值给error_list 一般第三个参数和第一个参数相同
#第四个参数，阻塞时间，如 1秒（这个如果不写，select会阻塞住，直到监听的描述符发生变化才继续往下执行）
readable_list, writeable_list, error_list = select.select(inputs,[],[],1) 

对于I/O多路复用，咱们上面的例子就可以了，但是为了遵循select规范需要把读和写进行分离：

#rlist -- wait until ready for reading  #等待直到有读的操作
#wlist -- wait until ready for writing  #等待直到有写的操作
#xlist -- wait for an ``exceptional condition'' #等待一个错误的情况

读和写他共享接收的数据，仅仅靠变量是完成不了的，还的需要借助外界的字典，字典里为每一个客户度维护了一个队列。收到的信息都放到队列了，然后返回的时候直接从队列里拿就可以了

Queue 队列

队列的特点：

1、队列是先进先出，栈是相反的，后进先出

2、队列是线程安全的

import Queue

q = Queue.Queue() #调用队列生成对象
q.put(1)  #存放第一个值到队列
q.put(2)  #存放第二个值到队列


print 'get frist one:',q.get() #获取队列的第一个值
print 'get second on:',q.get() #获取队列的第二个值

先进先出原则第一次存放的是1，第二次存放的是2，那么我们在获取值得时候，第一次获取的就是1，第二次就是2

看下面的例子如果队列里没有值怎么办？他会等待直到有数据为止：

q = Queue.Queue() #调用队列生成对象

q.put(1)  #存放第一个值到队列
q.put(2)  #存放第二个值到队列

a = q.get() #获取队列的第一个值
print 'get frist one:%s' % a
b = q.get() #获取队列的第二个值
print 'get second one:%s' % b
c = q.get()#获取队列的第三个值
print 'get third one:%s' % c

#结果:
'''
get frist one:1
get second one:2
#这里一直在等待着值进来~
'''

 如果不想让他等待，不管是否队列里都取数据，可以使用get_nowait，但是如果队列中没有数据就会报错

q = Queue.Queue() #调用队列生成对象

q.put(1)  #存放第一个值到队列
q.put(2)  #存放第二个值到队列

a = q.get() #获取队列的第一个值
print 'get frist one:%s' % a
b = q.get() #获取队列的第二个值
print 'get second one:%s' % b
c = q.get_nowait()#获取队列的第三个值 ，使用：get_nowait()
print 'get third one:%s' % c

如果队列为空的时候可以通过异常处理进行捕获：

q = Queue.Queue() #调用队列生成对象
try:
    q.get_nowait()
except Queue.Empty as f:
    print 'The Queue is empty!'

同样的如果队列长度为2，如果队列满了之后，同样他也是等待，直到有位置才会继续如下代码：

q = Queue.Queue(2) #调用队列生成对象


q.put(1)  #存放第一个值到队列
print 'put value 1 done'
q.put(2)  #存放第二个值到队列
print 'put vlaue 2 done'
q.put(3) #存放第三个值到队列
print 'put value 3 done'


#结果：
'''
put value 1 done
put vlaue 2 done
#这里会一直等待~
'''

同样如果存放数值的时候如果不想让他等待，使用put_nowait()但是队列无法存放后会报错！

q = Queue.Queue(2) #调用队列生成对象

q.put(1)  #存放第一个值到队列
print 'put value 1 done'
q.put(2)  #存放第二个值到队列
print 'put vlaue 2 done'
q.put_nowait(3) #存放第三个值到队列，如果使用put_nowait()队列无法存放后会报错！
print 'put value 3 done'
#结果：
'''
put value 1 done
put vlaue 2 done
#这里会一直等待~

 利用select模拟伪Socket Server操作实例并把读/写进行分离:

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import select
import socket
import Queue
import time

sk = socket.socket()
sk.bind(('127.0.0.1',8000))
sk.listen(5)
sk.setblocking(False)   #定义非阻塞
inputs = [sk,]  #定义一个列表，select第一个参数监听句柄序列,当有变动是，捕获并把socket server加入到句柄序列中
outputs = []    #定义一个列表，select第二个参数监听句柄序列，当有值时就捕获，并加入到句柄序列
message = {}
#message的样板信息
#message = {
#    'c1':队列，[这里存放着用户C1发过来的消息]例如:[message1,message2]
#    'c2':队列，[这里存放着用户C2发过来的消息]例如:[message1,message2]
#}


while True:
    readable_list, writeable_list, error_list = select.select(inputs,outputs,[],1)
    #文件描述符可读 readable_list    只有第一个参数变化时候才捕获，并赋值给readable_list
    #文件描述符可写 writeable_list   只要有值，第二个参数就捕获并赋值给writeable_list
    #time.sleep(2)
    print 'inputs:',inputs
    print 'output:'
    print 'readable_list:',readable_list
    print 'writeable_list:',writeable_list
    print 'message',message
    for r in readable_list:     #当readable_list有值得时候循环
        if r == sk:     #判断是否为链接请求变化的是否是socket server
            conn,addr = r.accept()  #获取请求
            inputs.append(conn)     #把客户端对象（句柄）加入到inputs里
            message[conn] = Queue.Queue()    #并在字典里为这个客户端连接建立一个消息队列
        else:
            client_data = r.recv(1024)  #如果请求的不是sk是客户端接收消息
            if client_data:     #如果有数据
                outputs.append(r)   #把用户加入到outpus里触发select第二个参数
                message[r].put(client_data)     #在指定队列中插入数据
            else:
                inputs.remove(r)    #没有数据，删除监听链接
                del message[r]      #当数据为空的时候删除队列~~
    for w in writeable_list:        #如果第二个参数有数据
        try:
            data = message[w].get_nowait()  #去指定队列取数据 并且不阻塞
            w.sendall(data)     #返回请求输入给client端
        except Queue.Empty:     #反之触发异常
            pass
        outputs.remove(w)       #因为第二个参数有值得时候就触发捕获值，所以使用完之后需要移除它
        #del message[r]
    print '%s' %('-' * 40)

使用select() 的事件驱动模型只用单线程（进程）执行，占用资源少，不消耗太多 CPU，同时能够为多客户端提供服务。如果试图建立一个简单的事件驱动的服务器程序，这个模型有一定的参考价值。
但这个模型依旧有着很多问题。

首先select()接口并不是实现“事件驱动”的最好选择。因为当需要探测的句柄值较大时，select()接口本身需要消耗大量时间去轮询各个句柄。很 多操作系统提供了更为高效的接口，如linux提供了epoll，BSD提供了queue，Solaris提供了/dev/poll，…。如果需要实现 更高效的服务器程序，类似epoll这样的接口更被推荐。遗憾的是不同的操作系统特供的epoll接口有很大差异，所以使用类似于epoll的接口实现具 有较好跨平台能力的服务器会比较困难。
其次，该模型将事件探测和事件响应夹杂在一起，一旦事件响应的执行体庞大，则对整个模型是灾难性的。

I/O多路复用的应用场景：

#（1）当客户处理多个描述字时（一般是交互式输入和网络套接口），必须使用I/O复用。
#（2）当一个客户同时处理多个套接口时，而这种情况是可能的，但很少出现。
#（3）如果一个TCP服务器既要处理监听套接口，又要处理已连接套接口，一般也要用到I/O复用。
#（4）如果一个服务器即要处理TCP，又要处理UDP，一般要使用I/O复用。
#（5）如果一个服务器要处理多个服务或多个协议，一般要使用I/O复用。
'''与多进程和多线程技术相比，I/O多路复用技术的最大优势是系统开销小，系统不必创建进程/线程，也不必维护这些进程/线程，从而大大减小了系统的开销。'''

 

参考文章：http://www.cnblogs.com/luotianshuai/p/5098408.html