Python全栈工程师（22：Socket编程11-Select解析Socket通信）

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

Windows Python：提供： select
Mac Python：提供： select
Linux Python：提供： select、poll、epoll

注意：网络操作、文件操作、终端操作等均属于IO操作，对于windows只支持Socket操作，其他系统支持其他IO操作，但是无法检测普通文件操作自动上次读取是否已经变化。

对于select方法：

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

#!/usr/bin/env python
# -*- coding:utf-8 -*-import select
import threading
import syswhile True:readable, writeable, error = select.select([sys.stdin,],[],[],1)if sys.stdin in readable:print 'select get stdin',sys.stdin.readline()

#!/usr/bin/env python
# -*- coding:utf-8 -*-import socket
import selectsk1 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sk1.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sk1.bind(('127.0.0.1',8002))
sk1.listen(5)
sk1.setblocking(0)inputs = [sk1,]while True:readable_list, writeable_list, error_list = select.select(inputs, [], inputs, 1)for r in readable_list:# 当客户端第一次连接服务端时if sk1 == r:print 'accept'request, address = r.accept()request.setblocking(0)inputs.append(request)# 当客户端连接上服务端之后，再次发送数据时else:received = r.recv(1024)# 当正常接收客户端发送的数据时if received:print 'received data:', received# 当客户端关闭程序时else:inputs.remove(r)sk1.close()

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import socketip_port = ('127.0.0.1',8002)
sk = socket.socket()
sk.connect(ip_port)while True:inp = raw_input('please input:')sk.sendall(inp)
sk.close()

此处的Socket服务端相比与原生的Socket，他支持当某一个请求不再发送数据时，服务器端不会等待而是可以去处理其他请求的数据。但是，如果每个请求的耗时比较长时，select版本的服务器端也无法完成同时操作。

#!/usr/bin/env python
#coding:utf8'''服务器的实现 采用select的方式
'''import select
import socket
import sys
import Queue#创建套接字并设置该套接字为非阻塞模式server = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
server.setblocking(0)#绑定套接字
server_address = ('localhost',10000)
print >>sys.stderr,'starting up on %s port %s'% server_address
server.bind(server_address)#将该socket变成服务模式
#backlog等于5，表示内核已经接到了连接请求，但服务器还没有调用accept进行处理的连接个数最大为5
#这个值不能无限大，因为要在内核中维护连接队列server.listen(5)#初始化读取数据的监听列表,最开始时希望从server这个套接字上读取数据
inputs = [server]#初始化写入数据的监听列表，最开始并没有客户端连接进来，所以列表为空outputs = []#要发往客户端的数据
message_queues = {}
while inputs:print >>sys.stderr,'waiting for the next event'#调用select监听所有监听列表中的套接字，并将准备好的套接字加入到对应的列表中readable,writable,exceptional = select.select(inputs,outputs,inputs)#列表中的socket 套接字  如果是文件呢？ #监控文件句柄有某一处发生了变化 可写 可读  异常属于Linux中的网络编程 #属于同步I/O操作，属于I/O复用模型的一种#rlist--等待到准备好读#wlist--等待到准备好写#xlist--等待到一种异常#处理可读取的套接字'''如果server这个套接字可读，则说明有新链接到来此时在server套接字上调用accept,生成一个与客户端通讯的套接字并将与客户端通讯的套接字加入inputs列表，下一次可以通过select检查连接是否可读然后在发往客户端的缓冲中加入一项，键名为:与客户端通讯的套接字，键值为空队列select系统调用是用来让我们的程序监视多个文件句柄(file descrīptor)的状态变化的。程序会停在select这里等待，直到被监视的文件句柄有某一个或多个发生了状态改变''''''若可读的套接字不是server套接字,有两种情况:一种是有数据到来，另一种是链接断开如果有数据到来,先接收数据,然后将收到的数据填入往客户端的缓存区中的对应位置，最后将于客户端通讯的套接字加入到写数据的监听列表:如果套接字可读.但没有接收到数据，则说明客户端已经断开。这时需要关闭与客户端连接的套接字进行资源清理'''for s in readable: if s is server:connection,client_address = s.accept()print >>sys.stderr,'connection from',client_addressconnection.setblocking(0)#设置非阻塞inputs.append(connection)message_queues[connection] = Queue.Queue()else:data = s.recv(1024)if data:print >>sys.stderr,'received "%s" from %s'% \(data,s.getpeername())message_queues[s].put(data)if s not in outputs:outputs.append(s)else:print >>sys.stderr,'closing',client_addressif s in outputs:outputs.remove(s)inputs.remove(s)s.close()del message_queues[s]#处理可写的套接字'''在发送缓冲区中取出响应的数据，发往客户端。如果没有数据需要写，则将套接字从发送队列中移除，select中不再监视'''for s in writable:try:next_msg = message_queues[s].get_nowait()except Queue.Empty:print >>sys.stderr,'  ',s,getpeername(),'queue empty'outputs.remove(s)else:print >>sys.stderr,'sending "%s" to %s'% \(next_msg,s.getpeername())s.send(next_msg)#处理异常情况for s in exceptional:for s in exceptional:print >>sys.stderr,'exception condition on',s.getpeername()inputs.remove(s)if s in outputs:outputs.remove(s)s.close()del message_queues[s]

实战案例：

服务器端：

#_*_coding:utf-8_*_import select
import socket
import sys
import queue# Create a TCP/IP socket
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setblocking(False)# Bind the socket to the port
server_address = ('localhost', 10000)
print(sys.stderr, 'starting up on %s port %s' % server_address)
server.bind(server_address)# Listen for incoming connections
server.listen(5)# Sockets from which we expect to read
inputs = [ server ]# Sockets to which we expect to write
outputs = [ ]message_queues = {}
while inputs:# Wait for at least one of the sockets to be ready for processingprint( '\nwaiting for the next event')readable, writable, exceptional = select.select(inputs, outputs, inputs)# Handle inputsfor s in readable:if s is server:# A "readable" server socket is ready to accept a connectionconnection, client_address = s.accept()print('new connection from', client_address)connection.setblocking(False)inputs.append(connection)# Give the connection a queue for data we want to sendmessage_queues[connection] = queue.Queue()else:data = s.recv(1024)if data:# A readable client socket has dataprint(sys.stderr, 'received "%s" from %s' % (data, s.getpeername()) )message_queues[s].put(data)# Add output channel for responseif s not in outputs:outputs.append(s)else:# Interpret empty result as closed connectionprint('closing', client_address, 'after reading no data')# Stop listening for input on the connectionif s in outputs:outputs.remove(s)  #既然客户端都断开了，我就不用再给它返回数据了，所以这时候如果这个客户端的连接对象还在outputs列表中，就把它删掉inputs.remove(s)    #inputs中也删除掉s.close()           #把这个连接关闭掉# Remove message queuedel message_queues[s]# Handle outputsfor s in writable:try:next_msg = message_queues[s].get_nowait()except queue.Empty:# No messages waiting so stop checking for writability.print('output queue for', s.getpeername(), 'is empty')outputs.remove(s)else:print( 'sending "%s" to %s' % (next_msg, s.getpeername()))s.send(next_msg)# Handle "exceptional conditions"for s in exceptional:print('handling exceptional condition for', s.getpeername() )# Stop listening for input on the connectioninputs.remove(s)if s in outputs:outputs.remove(s)s.close()# Remove message queuedel message_queues[s]

客户端：

# _*_coding:utf-8_*_
import socket
import sysmessages = ['This is the message. ','It will be sent ','in parts.',]
server_address = ('localhost', 10000)# Create a TCP/IP socket
socks = [socket.socket(socket.AF_INET, socket.SOCK_STREAM),socket.socket(socket.AF_INET, socket.SOCK_STREAM),]# Connect the socket to the port where the server is listening
print(sys.stderr, 'connecting to %s port %s' % server_address)
for s in socks:s.connect(server_address)for message in messages:# Send messages on both socketsfor s in socks:print(sys.stderr, '%s: sending "%s"' % (s.getsockname(), message))s.send(message)# Read responses on both socketsfor s in socks:data = s.recv(1024)print(sys.stderr, '%s: received "%s"' % (s.getsockname(), data))if not data:print(sys.stderr, 'closing socket', s.getsockname())s.close()

服务器端输出：

$ python ./select_echo_server.py
starting up on localhost port 10000waiting for the next event
new connection from ('127.0.0.1', 55821)waiting for the next event
new connection from ('127.0.0.1', 55822)
received "This is the message. " from ('127.0.0.1', 55821)waiting for the next event
sending "This is the message. " to ('127.0.0.1', 55821)waiting for the next event
output queue for ('127.0.0.1', 55821) is emptywaiting for the next event
received "This is the message. " from ('127.0.0.1', 55822)waiting for the next event
sending "This is the message. " to ('127.0.0.1', 55822)waiting for the next event
output queue for ('127.0.0.1', 55822) is emptywaiting for the next event
received "It will be sent " from ('127.0.0.1', 55821)
received "It will be sent " from ('127.0.0.1', 55822)waiting for the next event
sending "It will be sent " to ('127.0.0.1', 55821)
sending "It will be sent " to ('127.0.0.1', 55822)waiting for the next event
output queue for ('127.0.0.1', 55821) is empty
output queue for ('127.0.0.1', 55822) is emptywaiting for the next event
received "in parts." from ('127.0.0.1', 55821)
received "in parts." from ('127.0.0.1', 55822)waiting for the next event
sending "in parts." to ('127.0.0.1', 55821)
sending "in parts." to ('127.0.0.1', 55822)waiting for the next event
output queue for ('127.0.0.1', 55821) is empty
output queue for ('127.0.0.1', 55822) is emptywaiting for the next event
closing ('127.0.0.1', 55822) after reading no data
closing ('127.0.0.1', 55822) after reading no datawaiting for the next event

客户端输出：

$ python ./select_echo_multiclient.py
connecting to localhost port 10000
('127.0.0.1', 55821): sending "This is the message. "
('127.0.0.1', 55822): sending "This is the message. "
('127.0.0.1', 55821): received "This is the message. "
('127.0.0.1', 55822): received "This is the message. "
('127.0.0.1', 55821): sending "It will be sent "
('127.0.0.1', 55822): sending "It will be sent "
('127.0.0.1', 55821): received "It will be sent "
('127.0.0.1', 55822): received "It will be sent "
('127.0.0.1', 55821): sending "in parts."
('127.0.0.1', 55822): sending "in parts."
('127.0.0.1', 55821): received "in parts."
('127.0.0.1', 55822): received "in parts."

144-149

总结：

epoll linux ---libevent.so

gevent ---libevent.so

selectors

Twisted异步网络框架-- top3(代码量）学习成本高！【不是特别有必要】

http://www.cnblogs.com/alex3714/articles/5248247.html

Twsited异步网络框架

Twisted是一个事件驱动的网络框架，其中包含了诸多功能，例如：网络协议、线程、数据库管理、网络操作、电子邮件等。　

事件驱动

简而言之，事件驱动分为二个部分：第一，注册事件；第二，触发事件。

自定义事件驱动框架，命名为：“弑君者”：

#!/usr/bin/env python
# -*- coding:utf-8 -*-# event_drive.pyevent_list = []def run():for event in event_list:obj = event()obj.execute()class BaseHandler(object):"""用户必须继承该类，从而规范所有类的方法（类似于接口的功能）"""def execute(self):raise Exception('you must overwrite execute')最牛逼的事件驱动框架

程序员使用“弑君者框架”：　　

#!/usr/bin/env python
# -*- coding:utf-8 -*-from source import event_driveclass MyHandler(event_drive.BaseHandler):def execute(self):print 'event-drive execute MyHandler'event_drive.event_list.append(MyHandler)
event_drive.run()

Protocols

Protocols描述了如何以异步的方式处理网络中的事件。HTTP、DNS以及IMAP是应用层协议中的例子。Protocols实现了IProtocol接口，它包含如下的方法：

makeConnection               在transport对象和服务器之间建立一条连接
connectionMade               连接建立起来后调用
dataReceived                 接收数据时调用
connectionLost               关闭连接时调用

Transports

Transports代表网络中两个通信结点之间的连接。Transports负责描述连接的细节，比如连接是面向流式的还是面向数据报的，流控以及可靠性。TCP、UDP和Unix套接字可作为transports的例子。它们被设计为“满足最小功能单元，同时具有最大程度的可复用性”，而且从协议实现中分离出来，这让许多协议可以采用相同类型的传输。Transports实现了ITransports接口，它包含如下的方法：

write                   以非阻塞的方式按顺序依次将数据写到物理连接上
writeSequence           将一个字符串列表写到物理连接上
loseConnection          将所有挂起的数据写入，然后关闭连接
getPeer                 取得连接中对端的地址信息
getHost                 取得连接中本端的地址信息

将transports从协议中分离出来也使得对这两个层次的测试变得更加简单。可以通过简单地写入一个字符串来模拟传输，用这种方式来检查。

EchoServer

from twisted.internet import protocol
from twisted.internet import reactorclass Echo(protocol.Protocol):def dataReceived(self, data):self.transport.write(data)def main():factory = protocol.ServerFactory()factory.protocol = Echoreactor.listenTCP(1234,factory)reactor.run()if __name__ == '__main__':main()

EchoClient

from twisted.internet import reactor, protocol# a client protocolclass EchoClient(protocol.Protocol):"""Once connected, send a message, then print the result."""def connectionMade(self):self.transport.write("hello alex!")def dataReceived(self, data):"As soon as any data is received, write it back."print "Server said:", dataself.transport.loseConnection()def connectionLost(self, reason):print "connection lost"class EchoFactory(protocol.ClientFactory):protocol = EchoClientdef clientConnectionFailed(self, connector, reason):print "Connection failed - goodbye!"reactor.stop()def clientConnectionLost(self, connector, reason):print "Connection lost - goodbye!"reactor.stop()# this connects the protocol to a server running on port 8000
def main():f = EchoFactory()reactor.connectTCP("localhost", 1234, f)reactor.run()# this only runs if the module was *not* imported
if __name__ == '__main__':main()

运行服务器端脚本将启动一个TCP服务器，监听端口1234上的连接。服务器采用的是Echo协议，数据经TCP transport对象写出。运行客户端脚本将对服务器发起一个TCP连接，回显服务器端的回应然后终止连接并停止reactor事件循环。这里的Factory用来对连接的双方生成protocol对象实例。两端的通信是异步的，connectTCP负责注册回调函数到reactor事件循环中，当socket上有数据可读时通知回调处理。

一个传送文件的例子

server side

#_*_coding:utf-8_*_
# This is the Twisted Fast Poetry Server, version 1.0import optparse, osfrom twisted.internet.protocol import ServerFactory, Protocoldef parse_args():usage = """usage: %prog [options] poetry-fileThis is the Fast Poetry Server, Twisted edition.
Run it like this:python fastpoetry.py <path-to-poetry-file>If you are in the base directory of the twisted-intro package,
you could run it like this:python twisted-server-1/fastpoetry.py poetry/ecstasy.txtto serve up John Donne's Ecstasy, which I know you want to do.
"""parser = optparse.OptionParser(usage)help = "The port to listen on. Default to a random available port."parser.add_option('--port', type='int', help=help)help = "The interface to listen on. Default is localhost."parser.add_option('--iface', help=help, default='localhost')options, args = parser.parse_args()print("--arg:",options,args)if len(args) != 1:parser.error('Provide exactly one poetry file.')poetry_file = args[0]if not os.path.exists(args[0]):parser.error('No such file: %s' % poetry_file)return options, poetry_fileclass PoetryProtocol(Protocol):def connectionMade(self):self.transport.write(self.factory.poem)self.transport.loseConnection()class PoetryFactory(ServerFactory):protocol = PoetryProtocoldef __init__(self, poem):self.poem = poemdef main():options, poetry_file = parse_args()poem = open(poetry_file).read()factory = PoetryFactory(poem)from twisted.internet import reactorport = reactor.listenTCP(options.port or 9000, factory,interface=options.iface)print 'Serving %s on %s.' % (poetry_file, port.getHost())reactor.run()if __name__ == '__main__':main()

client side 　　

# This is the Twisted Get Poetry Now! client, version 3.0.# NOTE: This should not be used as the basis for production code.import optparsefrom twisted.internet.protocol import Protocol, ClientFactorydef parse_args():usage = """usage: %prog [options] [hostname]:port ...This is the Get Poetry Now! client, Twisted version 3.0
Run it like this:python get-poetry-1.py port1 port2 port3 ...
"""parser = optparse.OptionParser(usage)_, addresses = parser.parse_args()if not addresses:print parser.format_help()parser.exit()def parse_address(addr):if ':' not in addr:host = '127.0.0.1'port = addrelse:host, port = addr.split(':', 1)if not port.isdigit():parser.error('Ports must be integers.')return host, int(port)return map(parse_address, addresses)class PoetryProtocol(Protocol):poem = ''def dataReceived(self, data):self.poem += datadef connectionLost(self, reason):self.poemReceived(self.poem)def poemReceived(self, poem):self.factory.poem_finished(poem)class PoetryClientFactory(ClientFactory):protocol = PoetryProtocoldef __init__(self, callback):self.callback = callbackdef poem_finished(self, poem):self.callback(poem)def get_poetry(host, port, callback):"""Download a poem from the given host and port and invokecallback(poem)when the poem is complete."""from twisted.internet import reactorfactory = PoetryClientFactory(callback)reactor.connectTCP(host, port, factory)def poetry_main():addresses = parse_args()from twisted.internet import reactorpoems = []def got_poem(poem):poems.append(poem)if len(poems) == len(addresses):reactor.stop()for address in addresses:host, port = addressget_poetry(host, port, got_poem)reactor.run()for poem in poems:print poemif __name__ == '__main__':poetry_main()

Twisted深入

http://krondo.com/an-introduction-to-asynchronous-programming-and-twisted/

http://blog.csdn.net/hanhuili/article/details/9389433