Python网络编程
一：socket
网络中进程之间如何通信？
本地的进程间通信（IPC）有很多种方式，但可以总结为下面4类：
- 消息传递（管道、FIFO、消息队列）
- 同步（互斥量、条件变量、读写锁、文件和写记录锁、信号量）
- 共享内存（匿名的和具名的）
网络间的进程是如何通信的呢？
首要解决的问题是如何唯一标识一个进程，否则通信无从谈起！在本地可以通过进程PID来唯一标识一个进程，但是在网络中这是行不通的。其实TCP/IP协议族已经帮我们解决了这个问题，网络层的ip地址可以唯一标识网络中的主机，而传输层的协议+端口可以唯一标识主机中的进程。这样利用三元组（ip地址，协议，端口）就可以标识网络的进程了，网络中的进程通信就可以利用这个标志与其它进程进行交互。
什么是Socket？
上面我们已经知道网络中的进程是通过socket来通信的，那什么是socket呢？
socket起源于Unix，而Unix/Linux基本哲学之一就是“一切皆文件”，都可以用“打开open –> 读写write/read –> 关闭close”模式来操作。我的理解就是Socket就是该模式的一个实现，socket即是一种特殊的文件，一些socket函数就是对其进行的操作（读/写IO、打开、关闭
看图说话socket作用:
{% asset_img socket.png socket %}
如下图是socket的工作流程：
{% asset_img socket_connections.png socket_connections %}
上图是一张socket的tcp通信简图，我们都知道tcp的通信需要三次握手。tcp是可靠的、面向连接的、尽力传输的协议，而udp是不可靠的、面向非连接的、不尽力传输的协议。但是不可靠不代表它没有用，udp有自己的应用场景，语音和视频几乎都在使用udp协议，它的不可靠只是相对于tcp来说的，但是它的好处就是效率，高效在某些场景要比可靠性重要。这就涉及trade-off了，也就是权衡，需要根据你的应用权衡利弊，然后进行选择。
在socket选择初始化一个tcp协议的socket之后，就会绑定一个地址和端口，然后开始listen，客户端连接这个listen的tcp 之后，服务端会accept这个请求，然后产生一个新的socket，双方使用这个新的socket（地址和端口，地址还是上面listen的地址，端口 会是一个新的，这个从打印出的结果中可以看出）进行后续的通信。原来的端口会继续的listen新的请求。
server的步骤：
第一步: 创建socket对象。调用socket构造函数。如：
socket = socket.socket( family, type )
socket函数对应于普通文件的打开操作。普通文件的打开操作返回一个文件描述字，而socket()用于创建一个socket描述符（socket descriptor），它唯一标识一个socket。这个socket描述字跟文件描述字一样，后续的操作都有用到它，把它作为参数，通过它来进行一些读写操作。
正如可以给fopen的传入不同参数值，以打开不同的文件。创建socket的时候，也可以指定不同的参数创建不同的socket描述符，socket函数的三个参数分别为：
- domain：即协议域，又称为协议族（family）。
常用的协议族有，AF_INET、AF_INET6、AF_LOCAL（或称AF_UNIX，Unix域socket）、AF_ROUTE等等。协议族决定了socket的地址类型，在通信中必须采用对应的地址，如AF_INET决定了要用ipv4地址（32位的）与端口号（16位的）的组合、AF_UNIX决定了要用一个绝对路径名作为地址。
- type：指定socket类型。
常用的socket类型有，SOCK_STREAM、SOCK_DGRAM、SOCK_RAW、SOCK_PACKET、SOCK_SEQPACKET等等（socket的类型有哪些？）。 - protocol：故名思意，就是指定协议。
常用的协议有，IPPROTO_TCP、IPPTOTO_UDP、IPPROTO_SCTP、IPPROTO_TIPC等，它们分别对应TCP传输协议、UDP传输协议、STCP传输协议、TIPC传输协议（这个协议我将会单独开篇讨论！）。
注意：并不是上面的type和protocol可以随意组合的，如SOCK_STREAM不可以跟IPPROTO_UDP组合。当protocol为0时，会自动选择type类型对应的默认协议。
当我们调用socket创建一个socket时，返回的socket描述字它存在于协议族
（address family，AF_XXX）空间中，但没有一个具体的地址。如果想要给它赋值一个地址，就必须调用bind()函数，否则就当调用connect()、listen()时系统会自动随机分配一个端口。
第二步: 将socket绑定到指定地址。这是通过socket对象的bind方法来实现的：
socket.bind(address)
由AF_INET所创建的套接字，address地址必须是一个双元素元组，格式是(host,port)。host代表主机，port代表端口号。如果端口号正在使用、主机名不正确或端口已被保留，bind方法将引发socket.error异常。
第三步: 使用socket套接字的listen方法接收连接请求。
socket.listen(backlog)
backlog指定最多允许多少个客户连接到服务器。它的值至少为1。收到连接请求后，这些请求需要排队，如果队列满，就拒绝请求。
第四步: 服务器套接字通过socket的accept方法等待客户请求一个连接。
connection, address = socket.accept()
调用accept方法时，socket会时入“waiting”状态。客户请求连接时，方法建立连接并返回服务器。accept方法返回一个含有两个元素的 元组(connection,address)。第一个元素connection是新的socket对象，服务器必须通过它与客户通信；第二个元素 address是客户的Internet地址。
第五步: 处理阶段，服务器和客户端通过send和recv方法通信(传输数据)。服务器调用send，并采用字符串形式向客户发送信息。send方法返回已发送的字符个数。服务器使用recv方法从客户接收信息。调用recv时，服务器必须指定一个整数，它对应于可通过本次方法调用来接收的最大数据量。recv方法在接收数据时会进入“blocked”状态，最后返回一个字符 串，用它表示收到的数据。如果发送的数据量超过了recv所允许的，数据会被截短。多余的数据将缓冲于接收端。以后调用recv时，多余的数据会从缓冲区 删除(以及自上次调用recv以来，客户可能发送的其它任何数据)。
第六步: 传输结束，服务器调用socket的close方法关闭连接。
例：
import socket
ip_port = ('127.0.0.1',9999)
sk = socket.socket()
sk.bind(ip_port)
sk.listen(5)
while True:
print('server waiting...')
conn,addr = sk.accept()
client_data = conn.recv(1024)
print(client_data)
conn.sendall('不要回答,不要回答,不要回答')
conn.close()
client的步骤：
第一步: 创建一个socket以连接服务器：
socket = socket.socket( family, type )
第二步: 使用socket的connect方法连接服务器。对于AF_INET家族,连接格式如下：
socket.connect( (host,port) )
host代表服务器主机名或IP，port代表服务器进程所绑定的端口号。如连接成功，客户就可通过套接字与服务器通信，如果连接失败，会引发socket.error异常。
第三步: 处理阶段，客户和服务器将通过send方法和recv方法通信。
第四步: 传输结束，客户通过调用socket的close方法关闭连接。
例：
import socket
ip_port = ('127.0.0.1',9999)
sk = socket.socket()
sk.connect(ip_port)
sk.sendall('请求占领地球')
server_reply = sk.recv(1024)
print(server_reply)
sk.close()
UDP demo
# UDP Server
import socket
ip_port = ('127.0.0.1',9999)
sk = socket.socket(socket.AF_INET,socket.SOCK_DGRAM,0)
sk.bind(ip_port)
while True:
data = sk.recv(1024)
print(data)
# UDP Client
import socket
ip_port = ('127.0.0.1',9999)
sk = socket.socket(socket.AF_INET,socket.SOCK_DGRAM,0)
while True:
inp = raw_input('数据：').strip()
if inp == 'exit':
break
sk.sendto(inp,ip_port)
sk.close()
Web服务应用的例子
import socket
def handle_request(client):
buf = client.recv(1024)
client.send("HTTP/1.1 200 OK\r\n\r\n")
client.send("Hello, World")
def main():
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind(('localhost',8080))
sock.listen(5)
while True:
connection, address = sock.accept()
handle_request(connection)
connection.close()
if __name__ == '__main__':
main()
介绍一下常用API：
sk.bind(address)
sk.bind(address) 将套接字绑定到地址。address地址的格式取决于地址族。在AF_INET下，以元组（host,port）的形式表示地址。
sk.listen(backlog)
开始监听传入连接。backlog指定在拒绝连接之前，可以挂起的最大连接数量。backlog等于5，表示内核已经接到了连接请求，但服务器还没有调用accept进行处理的连接个数最大为5,这个值不能无限大，因为要在内核中维护连接队列
sk.setblocking(bool)
是否阻塞（默认True），如果设置False，那么accept和recv时一旦无数据，则报错。
sk.accept()
接受连接并返回（conn,address）,其中conn是新的套接字对象，可以用来接收和发送数据。address是连接客户端的地址。接收TCP 客户的连接（阻塞式）等待连接的到来
sk.connect(address)
连接到address处的套接字。一般，address的格式为元组（hostname,port）,如果连接出错，返回socket.error错误。
sk.connect_ex(address)
同上，只不过会有返回值，连接成功时返回 0 ，连接失败时候返回编码，例如：10061
sk.close()
关闭套接字
sk.recv(bufsize[,flag])
接受套接字的数据。数据以字符串形式返回，bufsize指定最多可以接收的数量。flag提供有关消息的其他信息，通常可以忽略。
sk.recvfrom(bufsize[.flag])
与recv()类似，但返回值是（data,address）。其中data是包含接收数据的字符串，address是发送数据的套接字地址。
sk.send(string[,flag])
将string中的数据发送到连接的套接字。返回值是要发送的字节数量，该数量可能小于string的字节大小。即：可能未将指定内容全部发送。
sk.sendall(string[,flag])
将string中的数据发送到连接的套接字，但在返回之前会尝试发送所有数据。成功返回None，失败则抛出异常。内部通过递归调用send，将所有内容发送出去。
sk.sendto(string[,flag],address)
将数据发送到套接字，address是形式为（ipaddr，port）的元组，指定远程地址。返回值是发送的字节数。该函数主要用于UDP协议。
sk.settimeout(timeout)
设置套接字操作的超时期，timeout是一个浮点数，单位是秒。值为None表示没有超时期。一般，超时期应该在刚创建套接字时设置，因为它们可能用于连接的操作（如 client 连接最多等待5s ）
sk.getpeername()
返回连接套接字的远程地址。返回值通常是元组（ipaddr,port）。
sk.getsockname()
返回套接字自己的地址。通常是一个元组(ipaddr,port)
sk.fileno()
套接字的文件描述符
二：I/O多路复用
I/O多路复用指：通过一种机制，可以监视多个描述符，一旦某个描述符就绪（一般是读就绪或者写就绪），能够通知程序进行相应的读写操作。
IO多路复用是指内核一旦发现进程指定的一个或者多个IO条件准备读取，它就通知该进程。IO多路复用适用如下场合：
- 当客户处理多个描述字时（一般是交互式输入和网络套接口），必须使用I/O复用。
- 当一个客户同时处理多个套接口时，而这种情况是可能的，但很少出现。
- 如果一个TCP服务器既要处理监听套接口，又要处理已连接套接口，一般也要用到I/O复用。
- 如果一个服务器即要处理TCP，又要处理UDP，一般要使用I/O复用。
- 如果一个服务器要处理多个服务或多个协议，一般要使用I/O复用。
与多进程和多线程技术相比，I/O多路复用技术的最大优势是系统开销小，系统不必创建进程/线程，也不必维护这些进程/线程，从而大大减小了系统的开销。
Linux
Linux中的 select，poll，epoll 都是IO多路复用的机制。
** select**
select最早于1983年出现在4.2BSD中，它通过一个select()系统调用来监视多个文件描述符的数组，当select()返回后，该数组中就绪的文件描述符便会被内核修改标志位，使得进程可以获得这些文件描述符从而进行后续的读写操作。
select目前几乎在所有的平台上支持，其良好跨平台支持也是它的一个优点，事实上从现在看来，这也是它所剩不多的优点之一。
select的一个缺点在于单个进程能够监视的文件描述符的数量存在最大限制，在Linux上一般为1024，不过可以通过修改宏定义甚至重新编译内核的方式提升这一限制。
另外，select()所维护的存储大量文件描述符的数据结构，随着文件描述符数量的增大，其复制的开销也线性增长。同时，由于网络响应时间的延迟使得大量TCP连接处于非活跃状态，但调用select()会对所有socket进行一次线性扫描，所以这也浪费了一定的开销。
** poll**
poll在1986年诞生于System V Release 3，它和select在本质上没有多大差别，但是poll没有最大文件描述符数量的限制。
poll和select同样存在一个缺点就是，包含大量文件描述符的数组被整体复制于用户态和内核的地址空间之间，而不论这些文件描述符是否就绪，它的开销随着文件描述符数量的增加而线性增大。
另外，select()和poll()将就绪的文件描述符告诉进程后，如果进程没有对其进行IO操作，那么下次调用select()和poll()的时候将再次报告这些文件描述符，所以它们一般不会丢失就绪的消息，这种方式称为水平触发（Level Triggered）。
** epoll**
直到Linux2.6才出现了由内核直接支持的实现方法，那就是epoll，它几乎具备了之前所说的一切优点，被公认为Linux2.6下性能最好的多路I/O就绪通知方法。
epoll可以同时支持水平触发和边缘触发（Edge Triggered，只告诉进程哪些文件描述符刚刚变为就绪状态，它只说一遍，如果我们没有采取行动，那么它将不会再次告知，这种方式称为边缘触发），理论上边缘触发的性能要更高一些，但是代码实现相当复杂。
epoll同样只告知那些就绪的文件描述符，而且当我们调用epoll_wait()获得就绪文件描述符时，返回的不是实际的描述符，而是一个代表就绪描述符数量的值，你只需要去epoll指定的一个数组中依次取得相应数量的文件描述符即可，这里也使用了内存映射（mmap）技术，这样便彻底省掉了这些文件描述符在系统调用时复制的开销。
另一个本质的改进在于epoll采用基于事件的就绪通知方式。在select/poll中，进程只有在调用一定的方法后，内核才对所有监视的文件描述符进行扫描，而epoll事先通过epoll_ctl()来注册一个文件描述符，一旦基于某个文件描述符就绪时，内核会采用类似callback的回调机制，迅速激活这个文件描述符，当进程调用epoll_wait()时便得到通知。
Python
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 sys
while True:
readable, writeable, error = select.select([sys.stdin,],[],[],1)
if sys.stdin in readable:
print 'select get stdin',sys.stdin.readline()
利用select监听终端操作实例
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import socket
import select
sk1 = 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()
利用select实现伪同时处理多个Socket客户端请求：服务端
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import socket
ip_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()
利用select实现伪同时处理多个Socket客户端请求：客户端
此处的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_address
connection.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_address
if 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]
基于select实现socket服务端
select模块（实现伪并发）
Python中有一个select模块，其中提供了：select、poll、epoll三个方法，分别调用系统的 select，poll，epoll 从而实现IO多路复用
select 模块
Windows Python：
提供： select
Mac Python：
提供： select
Linux Python：
提供： select、poll、epoll
select
select最早于1983年出现在4.2BSD中，它通过一个select()系统调用来监视多个文件描述符的数组，当select()返回后，该数组中就绪的文件描述符便会被内核修改标志位，使得进程可以获得这些文件描述符从而进行后续的读写操作。select目前几乎在所有的平台上支持，其良好跨平台支持也是它的一个优点，事实上从现在看来，这也是它所剩不多的优点之一。select的一个缺点在于单个进程能够监视的文件描述符的数量存在最大限制，在Linux上一般为1024，不过可以通过修改宏定义甚至重新编译内核的方式提升这一限制。另外，select()所维护的存储大量文件描述符的数据结构，随着文件描述符数量的增大，其复制的开销也线性增长。同时，由于网络响应时间的延迟使得大量TCP连接处于非活跃状态，但调用select()会对所有socket进行一次线性扫描，所以这也浪费了一定的开销。
poll
poll在1986年诞生于System V Release 3，它和select在本质上没有多大差别，但是poll没有最大文件描述符数量的限制。poll和select同样存在一个缺点就是，包含大量文件描述符的数组被整体复制于用户态和内核的地址空间之间，而不论这些文件描述符是否就绪，它的开销随着文件描述符数量的增加而线性增大。另外，select()和poll()将就绪的文件描述符告诉进程后，如果进程没有对其进行IO操作，那么下次调用select()和poll()的时候将再次报告这些文件描述符，所以它们一般不会丢失就绪的消息，这种方式称为水平触发（Level Triggered）。
epoll
直到Linux2.6才出现了由内核直接支持的实现方法，那就是epoll，它几乎具备了之前所说的一切优点，被公认为Linux2.6下性能最好的多路I/O就绪通知方法。epoll可以同时支持水平触发和边缘触发（Edge Triggered，只告诉进程哪些文件描述符刚刚变为就绪状态，它只说一遍，如果我们没有采取行动，那么它将不会再次告知，这种方式称为边缘触发），理论上边缘触发的性能要更高一些，但是代码实现相当复杂。epoll同样只告知那些就绪的文件描述符，而且当我们调用epoll_wait()获得就绪文件描述符时，返回的不是实际的描述符，而是一个代表就绪描述符数量的值，你只需要去epoll指定的一个数组中依次取得相应数量的文件描述符即可，这里也使用了内存映射（mmap）技术，这样便彻底省掉了这些文件描述符在系统调用时复制的开销。另一个本质的改进在于epoll采用基于事件的就绪通知方式。在select/poll中，进程只有在调用一定的方法后，内核才对所有监视的文件描述符进行扫描，而epoll事先通过epoll_ctl()来注册一个文件描述符，一旦基于某个文件描述符就绪时，内核会采用类似callback的回调机制，迅速激活这个文件描述符，当进程调用epoll_wait()时便得到通知。
** select.select方法：**
select函数需要3个序列作为它的必选参数，此外还有一个可选的以秒单位的超时时间作为第4个参数。3个序列用于输入、输出以及异常情况（错误）；如果没有给定超时时间，select会阻塞（也就是处于等待状态），知道其中的一个文件描述符以及为行动做好了准备，如果给定了超时时间，select最多阻塞给定的超时时间，如果超时时间为0，那么就给出一个连续的poll（即不阻塞）；select的返回值是3个序列，每个代表相应参数的一个活动子集。第一个序列用于监听socket对象内部是否发生变化，如果有变化表示有新的连接，下面直接看程序代码
select.select伪并发程序服务端
import socket
import select
sk = socket.socket()
sk.bind(('127.0.0.1',8002))
sk.listen(5)
sk.setblocking(0) #不阻塞
inputs = [sk,]
messages = {}
outputs = []
while True:
readable_list, writeable_list, error_list = select.select(inputs, outputs, [], 1)
# readable_list 监听服务端对象，当inputs列表有变化时，变化的值会赋值给readable_list中
# 如果有新的连接进来，sk会发生变化，此时readable_list—的值为sk
# 如果conn对象发生变化，表示客户端发送了新的消息过来，此时readable_list的值为客户端连接
# writeable_lists实现读写分离，需要回复信息的conn对象添加到里面
print(len(inputs),len(readable_list),len(writeable_list),len(outputs))
for r in readable_list:
# 当客户端第一次连接服务端时，未在inputs里
if r == sk:
print('accept')
conn, address = r.accept()
conn.sendall("hello".encode())
inputs.append(conn) #添加到inputs
messages[conn]=[] #设置messages key值r为列表
# 当客户端连接上服务端之后，再次发送数据时，已经存在inputs
else:
try:
received = r.recv(1024)
# 当正常接收客户端发送的数据时
if not received:
raise Exception("断开连接")
else:
messages[r].append(received)
outputs.append(r)
# 当客户端关闭程序时
except Exception as e:
inputs.remove(r)
del messages[r]
for w in writeable_list:
msg = messages[w].pop()
rest = msg + "response".encode()
w.sendall(rest)
outputs.remove(w)
sk.close()
select.select伪并发程序客户端
import socket
sk = socket.socket()
sk.connect(("127.0.0.1",8002))
print(sk.recv(1024).decode())
while True:
command = input("--->>>")
sk.sendall(command.encode())
res = sk.recv(1024)
print(res.decode())
sk.close()
select.poll方法：
poll方法使用起来比select简单。在调用poll时，会得到一个poll对象。然后就可以使用poll的对象的register方法注册一个文件描述符（或者是带有fileno方法的对象）。注册后可以使用unregister方法移出注册的对象。注册了一些对象（比如套接字）以后，就可以调用poll方法（带有一个可选的超时时间参数）并得到一个（fd，event）格式列表（可能为空），其中fd是文件描述符，event则告诉你发生了什么。这是一个位掩码（bitmask），意思是它是一个整数，这个整数的每个位对应不同的事件。那些不同的事件是select模块的常量，为了验证是否设置了一个定位（也就是说，一个给定的事件是否发生了），可以使用按位与操作符（&）：if event & select.POLLIN
select模块中的polling事件常量：
事件名 描述
POLLIN 读取来自文件描述符的数据
POLLPRT 读取来自文件描述符的紧急数据
POLLOUT 文件描述符已经准备好数据，写入时不会发生阻塞
POLLERR 与文件描述符有关的错误情况
POLLHUP 挂起，连接丢失
POLLNVAL 无效请求，连接没有打开
poll的简单程序服务端（linux）
#poll 异步I/O
import socket,select
s = socket.socket()
host = "127.0.0.1"
port = 8002
s.bind((host,port))
fdmap = {s.fileno():s} #文件描述符到套接字对象的映射
s.listen(5)
p = select.poll() #poll对象
p.register(s) #注册一个文件描述符（带有fileno方法的对象)
while True:
events = p.poll()
for fd,event in events:
if fd == s.fileno(): #新的连接进来
c,addr = s.accept()
print("Got connectins from",addr)
p.register(c) #注册一个文件描述符（带有fileno方法的对象)
fdmap[c.fileno()] = c #添加到fdmap
elif event & select.POLLIN: #读取来自文件描述符的数据
data = fdmap[fd].recv(1024)
if not data: #表示客户端断开
print(fdmap[fd].getpeername(),"disconnected")
p.unregister(fd) #清除文件描述符
del fdmap[fd] #删除fdmap对应的key值
else:
print(data.decode())
poll程序客户端
#poll 异步I/O
import socket
sk = socket.socket()
sk.connect(("127.0.0.1",8002))
while True:
command = input("--->>>")
sk.sendall(command.encode())
sk.close()
** epoll方法：**
epoll是在2.6内核中提出的，是之前的select和poll的增强版本。相对于select和poll来说，epoll更加灵活，没有描述符限制。epoll使用一个文件描述符管理多个描述符，将用户关系的文件描述符的事件存放到内核的一个事件表中，这样在用户空间和内核空间的copy只需一次。
一 epoll操作过程
epoll操作过程需要三个接口，分别如下：
int epoll_create(int size)；//创建一个epoll的句柄，size用来告诉内核这个监听的数目一共有多大
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event)；
int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);
-
int epoll_create(int size);
创建一个epoll的句柄，size用来告诉内核这个监听的数目一共有多大，这个参数不同于select()中的第一个参数，给出最大监听的fd+1的值，参数size并不是限制了epoll所能监听的描述符最大个数，只是对内核初始分配内部数据结构的一个建议。
当创建好epoll句柄后，它就会占用一个fd值，在linux下如果查看/proc/进程id/fd/，是能够看到这个fd的，所以在使用完epoll后，必须调用close()关闭，否则可能导致fd被耗尽。 -
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event)；函数是对指定描述符fd执行op操作。
-epfd：是epoll_create()的返回值。
-op：表示op操作，用三个宏来表示：添加EPOLL_CTL_ADD，删除EPOLL_CTL_DEL，修改EPOLL_CTL_MOD。分别添加、删除和修改对fd的监听事件。
-fd：是需要监听的fd（文件描述符）
-epoll_event：是告诉内核需要监听什么事 -
int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);
等待epfd上的io事件，最多返回maxevents个事件。
参数events用来从内核得到事件的集合，maxevents告之内核这个events有多大，这个maxevents的值不能大于创建epoll_create()时的size，参数timeout是超时时间（毫秒，0会立即返回，-1将不确定，也有说法说是永久阻塞）。该函数返回需要处理的事件数目，如返回0表示已超时。
#_*_coding:utf-8_*_
__author__ = 'Alex Li'
import socket, logging
import select, errno
logger = logging.getLogger("network-server")
def InitLog():
logger.setLevel(logging.DEBUG)
fh = logging.FileHandler("network-server.log")
fh.setLevel(logging.DEBUG)
ch = logging.StreamHandler()
ch.setLevel(logging.ERROR)
formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
ch.setFormatter(formatter)
fh.setFormatter(formatter)
logger.addHandler(fh)
logger.addHandler(ch)
if __name__ == "__main__":
InitLog()
try:
# 创建 TCP socket 作为监听 socket
listen_fd = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
except socket.error as msg:
logger.error("create socket failed")
try:
# 设置 SO_REUSEADDR 选项
listen_fd.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
except socket.error as msg:
logger.error("setsocketopt SO_REUSEADDR failed")
try:
# 进行 bind -- 此处未指定 ip 地址，即 bind 了全部网卡 ip 上
listen_fd.bind(('', 2003))
except socket.error as msg:
logger.error("bind failed")
try:
# 设置 listen 的 backlog 数
listen_fd.listen(10)
except socket.error as msg:
logger.error(msg)
try:
# 创建 epoll 句柄
epoll_fd = select.epoll()
# 向 epoll 句柄中注册 监听 socket 的 可读 事件
epoll_fd.register(listen_fd.fileno(), select.EPOLLIN)
except select.error as msg:
logger.error(msg)
connections = {}
addresses = {}
datalist = {}
while True:
# epoll 进行 fd 扫描的地方 -- 未指定超时时间则为阻塞等待
epoll_list = epoll_fd.poll()
for fd, events in epoll_list:
# 若为监听 fd 被激活
if fd == listen_fd.fileno():
# 进行 accept -- 获得连接上来 client 的 ip 和 port，以及 socket 句柄
conn, addr = listen_fd.accept()
logger.debug("accept connection from %s, %d, fd = %d" % (addr[0], addr[1], conn.fileno()))
# 将连接 socket 设置为 非阻塞
conn.setblocking(0)
# 向 epoll 句柄中注册 连接 socket 的 可读 事件
epoll_fd.register(conn.fileno(), select.EPOLLIN | select.EPOLLET)
# 将 conn 和 addr 信息分别保存起来
connections[conn.fileno()] = conn
addresses[conn.fileno()] = addr
elif select.EPOLLIN & events:
# 有 可读 事件激活
datas = ''
while True:
try:
# 从激活 fd 上 recv 10 字节数据
data = connections[fd].recv(10)
# 若当前没有接收到数据，并且之前的累计数据也没有
if not data and not datas:
# 从 epoll 句柄中移除该 连接 fd
epoll_fd.unregister(fd)
# server 侧主动关闭该 连接 fd
connections[fd].close()
logger.debug("%s, %d closed" % (addresses[fd][0], addresses[fd][1]))
break
else:
# 将接收到的数据拼接保存在 datas 中
datas += data
except socket.error as msg:
# 在 非阻塞 socket 上进行 recv 需要处理 读穿 的情况
# 这里实际上是利用 读穿 出 异常 的方式跳到这里进行后续处理
if msg.errno == errno.EAGAIN:
logger.debug("%s receive %s" % (fd, datas))
# 将已接收数据保存起来
datalist[fd] = datas
# 更新 epoll 句柄中连接d 注册事件为 可写
epoll_fd.modify(fd, select.EPOLLET | select.EPOLLOUT)
break
else:
# 出错处理
epoll_fd.unregister(fd)
connections[fd].close()
logger.error(msg)
break
elif select.EPOLLHUP & events:
# 有 HUP 事件激活
epoll_fd.unregister(fd)
connections[fd].close()
logger.debug("%s, %d closed" % (addresses[fd][0], addresses[fd][1]))
elif select.EPOLLOUT & events:
# 有 可写 事件激活
sendLen = 0
# 通过 while 循环确保将 buf 中的数据全部发送出去
while True:
# 将之前收到的数据发回 client -- 通过 sendLen 来控制发送位置
sendLen += connections[fd].send(datalist[fd][sendLen:])
# 在全部发送完毕后退出 while 循环
if sendLen == len(datalist[fd]):
break
# 更新 epoll 句柄中连接 fd 注册事件为 可读
epoll_fd.modify(fd, select.EPOLLIN | select.EPOLLET)
else:
# 其他 epoll 事件不进行处理
continue
# epoll socket echo server
selectors模块
selectors模块已经封装了epoll，select方法；epoll优先级大于select
import selectors
import socket
sel = selectors.DefaultSelector()
def accept(sock, mask):
conn, addr = sock.accept() # Should be ready
print('accepted', conn, 'from', addr)
conn.setblocking(False)
sel.register(conn, selectors.EVENT_READ, read)
def read(conn, mask):
data = conn.recv(1000) # Should be ready
if data:
print('echoing', repr(data), 'to', conn)
conn.send(data) # Hope it won't block
else:
print('closing', conn)
sel.unregister(conn)
conn.close()
sock = socket.socket()
sock.bind(('localhost', 10000))
sock.listen(100)
sock.setblocking(False)
sel.register(sock, selectors.EVENT_READ, accept)
while True:
events = sel.select()
for key, mask in events:
callback = key.data
callback(key.fileobj, mask)
