IO多路复用,网络编制程序

本章内容:

Socket

socket平日也称得上”套接字”,用于描述IP地址和端口,是三个通讯链的句柄,应用程序平日经过”套接字”向网络发出诉求可能应答网络需要。

socket源点于Unix,而Unix/Linux基本法学之大器晚成正是“一切皆文件”,对于文本用【展开】【读写】【关闭】格局来操作。socket就是该格局的三个兑现,socket便是后生可畏种奇特的文书,一些socket函数正是对其开展的操作(读/写IO、张开、关闭)

socket和file的区别:

  • file模块是针对性某些钦点文件实行【打开】【读写】【关闭】
  • socket模块是本着 服务器端 和 客商端Socket
    进行【展开】【读写】【关闭】

图片 1

图片 2图片 3

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

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()

socket server

图片 4图片 5

#!/usr/bin/env python
# -*- coding:utf-8 -*-
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()

socket client

WEB服务应用:

#!/usr/bin/env python
#coding:utf-8
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()

越来越多效益

sk = socket.socket(socket.AF_INET,socket.SOCK_STREAM,0)

参数一:地址簇

  socket.AF_INET IPv4(默认)
  socket.AF_INET6 IPv6

  socket.AF_UNIX 只可以够用于单意气风发的Unix系统经过间通讯

参数二:类型

  socket.SOCK_STREAM  流式socket , for TCP (默认)
  socket.SOCK_DGRAM   数据报式socket , for UDP

  socket.SOCK_RAW
原始套接字,普通的套接字不也许管理ICMP、I卡那霉素P等互联网报文,而SOCK_RAW可以;其次,SOCK_RAW也得以管理非常的IPv4报文;其它,利用原始套接字,能够透过IP_HD奥德赛INCL套接字选项由顾客构造IP头。
  socket.SOCK_普拉多DM
是风度翩翩种保障的UDP格局,即确定保障交到数据报但不有限支撑顺序。SOCK_RAM用来提供对原本合同的最少访谈,在急需奉行某个特殊操作时行使,如发送ICMP报文。SOCK_RAM平日只限于高端客户或领队运行的次第接纳。
  socket.SOCK_SEQPACKET 可信赖的接连数据包服务

参数三:协议

  0  (暗许)与一定的地点家族有关的商业事务,借使是 0
,则系统就能够总部址格式和套接体系,自动选取三个适龄的磋商

图片 6图片 7

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

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()

UDP Demo

sk.bind(address)

  s.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()

  套接字的文书陈述符

图片 8图片 9

# 服务端
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,(host,port) = sk.recvfrom(1024)
    print(data,host,port)
    sk.sendto(bytes('ok', encoding='utf-8'), (host,port))


#客户端
import socket
ip_port = ('127.0.0.1',9999)

sk = socket.socket(socket.AF_INET,socket.SOCK_DGRAM,0)
while True:
    inp = input('数据:').strip()
    if inp == 'exit':
        break
    sk.sendto(bytes(inp, encoding='utf-8'),ip_port)
    data = sk.recvfrom(1024)
    print(data)

sk.close()

UDP

实例:智能机器人

图片 10图片 11

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


import socket

ip_port = ('127.0.0.1',8888)
sk = socket.socket()
sk.bind(ip_port)
sk.listen(5)

while True:
    conn,address =  sk.accept()
    conn.sendall('欢迎致电 10086,请输入1xxx,0转人工服务.')
    Flag = True
    while Flag:
        data = conn.recv(1024)
        if data == 'exit':
            Flag = False
        elif data == '0':
            conn.sendall('通过可能会被录音.balabala一大推')
        else:
            conn.sendall('请重新输入.')
    conn.close()

服务端

图片 12图片 13

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

import socket


ip_port = ('127.0.0.1',8005)
sk = socket.socket()
sk.connect(ip_port)
sk.settimeout(5)

while True:
    data = sk.recv(1024)
    print 'receive:',data
    inp = raw_input('please input:')
    sk.sendall(inp)
    if inp == 'exit':
        break

sk.close()

客户端

  • Socket
  • IO多路复用(select)
  • SocketServer
    模块(ThreadingTCPServer源码分析)

 

 

IO多路复用

I/O多路复用指:通过风华正茂种体制,能够监视五个描述符,风姿罗曼蒂克旦某些描述符就绪(日常是读就绪可能写就绪),可以布告顺序开展对应的读写操作。

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()时便得到通知。

Socket

Python

Python中有一个select模块,此中提供了:select、poll、epoll五个艺术,分别调用系统的 select,poll,epoll
从而实现IO多路复用。

Windows Python:

``提供: select

Mac Python:

``提供: select

Linux Python:

``提供: select、poll、epoll

瞩目:互连网操作、文件操作、终端操作等均属于IO操作,对于windows只协助Select
操作,别的系统扶持其余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``秒,之后返回三个空列表,如果监听的句柄有变化,则直接执行。

图片 14图片 15

#!/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监听终端操作实例

图片 16图片 17

#!/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顾客端乞求:服务端

图片 18图片 19

#!/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版本的劳务器端也无法成功并且操作。

图片 20图片 21

#!/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服务端(python2)

图片 22图片 23

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

sk1 = socket.socket()
sk1.bind(('127.0.0.1',8001))
sk1.listen()

inputs = [sk1, ]
# 实现读写分离
# 保存谁发送过消息
outputs = []
# 保存消息内容  可以用队列优化
message_dict = {}

while True:
    # select 内部自动监听多个句柄,一旦某个句柄发生变化则会响应
    # r_list变化值, w_list不变值, e_list出错值
    # 难点:inputs里面放着两类socket,一类是server的socket,一类是client的socket
    r_list, w_list, e_list = select.select(inputs,outputs,inputs,1)
    print('正在监听的socket对象%d' % len(inputs))
    print(r_list)
    for sk1_or_conn in r_list:
        # 每连接对象
        if sk1_or_conn == sk1:
            conn, address = sk1.accept()
            inputs.append(conn)
            # {'小明':[]}
            message_dict[conn] = []
        else:
            # 有老用户发消息了
            try:
                data_bytes = sk1_or_conn.recv(1024)
            except Exception as ex:
                inputs.remove(sk1_or_conn)
            else:
                # 用户正常发送消息
                data_str = str(data_bytes,encoding='utf-8')
                message_dict[sk1_or_conn].append(data_str)
                outputs.append(sk1_or_conn)

    # w_list 仅仅保存了谁给我发送过消息
    # w_list = [zhanghui,]
    for conn in w_list:
        recv_str = message_dict[conn][0]
        del message_dict[conn][0]
        conn.sendall(bytes(recv_str+"hao",encoding="utf-8"))
        outputs.remove(conn)

python3实现

 

socket常常也叫做”套接字”,用于描述IP地址和端口,是一个通信链的句柄,应用程序平日通过”套接字”向网络发出央浼只怕应答互联网诉求。

SocketServer模块

SocketServer内部运用 IO多路复用 甚至 “多线程” 和 “多进度”
,进而实现产出管理三个客商端央求的Socket服务端。即:各样客户端诉求连接到服务器时,Socket服务端都会在服务器是开创一个“线程”只怕“进度”
专责管理当下客商端的富有乞请。

图片 24

ThreadingTCPServer

ThreadingTCPServer达成的Soket服务器内部会为每个client创设一个 “线程”,该线程用来和客商端进行相互。

1、ThreadingTCPServer基础

使用ThreadingTCPServer:

  • 开创五个接续自 SocketServer.BaseRequestHandler 的类
  • 类中必得定义三个名号为 handle 的章程
  • 启动ThreadingTCPServer

图片 25图片 26

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

class MyServer(SocketServer.BaseRequestHandler):

    def handle(self):
        # print self.request,self.client_address,self.server
        conn = self.request
        conn.sendall('欢迎致电 10086,请输入1xxx,0转人工服务.')
        Flag = True
        while Flag:
            data = conn.recv(1024)
            if data == 'exit':
                Flag = False
            elif data == '0':
                conn.sendall('通过可能会被录音.balabala一大推')
            else:
                conn.sendall('请重新输入.')


if __name__ == '__main__':
    server = SocketServer.ThreadingTCPServer(('127.0.0.1',8009),MyServer)
    server.serve_forever()

SocketServer达成服务器

图片 27图片 28

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

import socket


ip_port = ('127.0.0.1',8009)
sk = socket.socket()
sk.connect(ip_port)
sk.settimeout(5)

while True:
    data = sk.recv(1024)
    print 'receive:',data
    inp = raw_input('please input:')
    sk.sendall(inp)
    if inp == 'exit':
        break

sk.close()

客户端

2、ThreadingTCPServer源码剖判

ThreadingTCPServer的类图关系如下:

图片 29 

个中调用流程为:

  • 起步服务端程序
  • 执行 TCPServer.__init__ 方法,创立服务端Socket对象并绑定 IP 和
    端口
  • 执行 BaseServer.__init__
    方法,将自定义的接续自SocketServer.BaseRequestHandler 的类
    MyRequestHandle赋值给 self.RequestHandlerClass
  • 执行 BaseServer.server_forever 方法,While
    循环一向监听是还是不是有客户端需要达到 …
  • 当客商端连接达到服务器
  • 执行 ThreadingMixIn.process_request 方法,创立三个 “线程”
    用来管理诉求
  • 执行 ThreadingMixIn.process_request_thread 方法
  • 执行 BaseServer.finish_request 方法,施行 self.RequestHandlerClass()
     即:施行自定义 MyRequestHandler
    的构造方法(自动调用基类BaseRequestHandler的构造方法,在该构造方法中又会调用
    MyRequestHandler的handle方法)

ThreadingTCPServer相关源码:

图片 30图片 31

class BaseServer:

    """Base class for server classes.

    Methods for the caller:

    - __init__(server_address, RequestHandlerClass)
    - serve_forever(poll_interval=0.5)
    - shutdown()
    - handle_request()  # if you do not use serve_forever()
    - fileno() -> int   # for select()

    Methods that may be overridden:

    - server_bind()
    - server_activate()
    - get_request() -> request, client_address
    - handle_timeout()
    - verify_request(request, client_address)
    - server_close()
    - process_request(request, client_address)
    - shutdown_request(request)
    - close_request(request)
    - handle_error()

    Methods for derived classes:

    - finish_request(request, client_address)

    Class variables that may be overridden by derived classes or
    instances:

    - timeout
    - address_family
    - socket_type
    - allow_reuse_address

    Instance variables:

    - RequestHandlerClass
    - socket

    """

    timeout = None

    def __init__(self, server_address, RequestHandlerClass):
        """Constructor.  May be extended, do not override."""
        self.server_address = server_address
        self.RequestHandlerClass = RequestHandlerClass
        self.__is_shut_down = threading.Event()
        self.__shutdown_request = False

    def server_activate(self):
        """Called by constructor to activate the server.

        May be overridden.

        """
        pass

    def serve_forever(self, poll_interval=0.5):
        """Handle one request at a time until shutdown.

        Polls for shutdown every poll_interval seconds. Ignores
        self.timeout. If you need to do periodic tasks, do them in
        another thread.
        """
        self.__is_shut_down.clear()
        try:
            while not self.__shutdown_request:
                # XXX: Consider using another file descriptor or
                # connecting to the socket to wake this up instead of
                # polling. Polling reduces our responsiveness to a
                # shutdown request and wastes cpu at all other times.
                r, w, e = _eintr_retry(select.select, [self], [], [],
                                       poll_interval)
                if self in r:
                    self._handle_request_noblock()
        finally:
            self.__shutdown_request = False
            self.__is_shut_down.set()

    def shutdown(self):
        """Stops the serve_forever loop.

        Blocks until the loop has finished. This must be called while
        serve_forever() is running in another thread, or it will
        deadlock.
        """
        self.__shutdown_request = True
        self.__is_shut_down.wait()

    # The distinction between handling, getting, processing and
    # finishing a request is fairly arbitrary.  Remember:
    #
    # - handle_request() is the top-level call.  It calls
    #   select, get_request(), verify_request() and process_request()
    # - get_request() is different for stream or datagram sockets
    # - process_request() is the place that may fork a new process
    #   or create a new thread to finish the request
    # - finish_request() instantiates the request handler class;
    #   this constructor will handle the request all by itself

    def handle_request(self):
        """Handle one request, possibly blocking.

        Respects self.timeout.
        """
        # Support people who used socket.settimeout() to escape
        # handle_request before self.timeout was available.
        timeout = self.socket.gettimeout()
        if timeout is None:
            timeout = self.timeout
        elif self.timeout is not None:
            timeout = min(timeout, self.timeout)
        fd_sets = _eintr_retry(select.select, [self], [], [], timeout)
        if not fd_sets[0]:
            self.handle_timeout()
            return
        self._handle_request_noblock()

    def _handle_request_noblock(self):
        """Handle one request, without blocking.

        I assume that select.select has returned that the socket is
        readable before this function was called, so there should be
        no risk of blocking in get_request().
        """
        try:
            request, client_address = self.get_request()
        except socket.error:
            return
        if self.verify_request(request, client_address):
            try:
                self.process_request(request, client_address)
            except:
                self.handle_error(request, client_address)
                self.shutdown_request(request)

    def handle_timeout(self):
        """Called if no new request arrives within self.timeout.

        Overridden by ForkingMixIn.
        """
        pass

    def verify_request(self, request, client_address):
        """Verify the request.  May be overridden.

        Return True if we should proceed with this request.

        """
        return True

    def process_request(self, request, client_address):
        """Call finish_request.

        Overridden by ForkingMixIn and ThreadingMixIn.

        """
        self.finish_request(request, client_address)
        self.shutdown_request(request)

    def server_close(self):
        """Called to clean-up the server.

        May be overridden.

        """
        pass

    def finish_request(self, request, client_address):
        """Finish one request by instantiating RequestHandlerClass."""
        self.RequestHandlerClass(request, client_address, self)

    def shutdown_request(self, request):
        """Called to shutdown and close an individual request."""
        self.close_request(request)

    def close_request(self, request):
        """Called to clean up an individual request."""
        pass

    def handle_error(self, request, client_address):
        """Handle an error gracefully.  May be overridden.

        The default is to print a traceback and continue.

        """
        print '-'*40
        print 'Exception happened during processing of request from',
        print client_address
        import traceback
        traceback.print_exc() # XXX But this goes to stderr!
        print '-'*40

BaseServer

图片 32图片 33

class TCPServer(BaseServer):

    """Base class for various socket-based server classes.

    Defaults to synchronous IP stream (i.e., TCP).

    Methods for the caller:

    - __init__(server_address, RequestHandlerClass, bind_and_activate=True)
    - serve_forever(poll_interval=0.5)
    - shutdown()
    - handle_request()  # if you don't use serve_forever()
    - fileno() -> int   # for select()

    Methods that may be overridden:

    - server_bind()
    - server_activate()
    - get_request() -> request, client_address
    - handle_timeout()
    - verify_request(request, client_address)
    - process_request(request, client_address)
    - shutdown_request(request)
    - close_request(request)
    - handle_error()

    Methods for derived classes:

    - finish_request(request, client_address)

    Class variables that may be overridden by derived classes or
    instances:

    - timeout
    - address_family
    - socket_type
    - request_queue_size (only for stream sockets)
    - allow_reuse_address

    Instance variables:

    - server_address
    - RequestHandlerClass
    - socket

    """

    address_family = socket.AF_INET

    socket_type = socket.SOCK_STREAM

    request_queue_size = 5

    allow_reuse_address = False

    def __init__(self, server_address, RequestHandlerClass, bind_and_activate=True):
        """Constructor.  May be extended, do not override."""
        BaseServer.__init__(self, server_address, RequestHandlerClass)
        self.socket = socket.socket(self.address_family,
                                    self.socket_type)
        if bind_and_activate:
            try:
                self.server_bind()
                self.server_activate()
            except:
                self.server_close()
                raise

    def server_bind(self):
        """Called by constructor to bind the socket.

        May be overridden.

        """
        if self.allow_reuse_address:
            self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.socket.bind(self.server_address)
        self.server_address = self.socket.getsockname()

    def server_activate(self):
        """Called by constructor to activate the server.

        May be overridden.

        """
        self.socket.listen(self.request_queue_size)

    def server_close(self):
        """Called to clean-up the server.

        May be overridden.

        """
        self.socket.close()

    def fileno(self):
        """Return socket file number.

        Interface required by select().

        """
        return self.socket.fileno()

    def get_request(self):
        """Get the request and client address from the socket.

        May be overridden.

        """
        return self.socket.accept()

    def shutdown_request(self, request):
        """Called to shutdown and close an individual request."""
        try:
            #explicitly shutdown.  socket.close() merely releases
            #the socket and waits for GC to perform the actual close.
            request.shutdown(socket.SHUT_WR)
        except socket.error:
            pass #some platforms may raise ENOTCONN here
        self.close_request(request)

    def close_request(self, request):
        """Called to clean up an individual request."""
        request.close()

TCPServer

图片 34图片 35

class ThreadingMixIn:
    """Mix-in class to handle each request in a new thread."""

    # Decides how threads will act upon termination of the
    # main process
    daemon_threads = False

    def process_request_thread(self, request, client_address):
        """Same as in BaseServer but as a thread.

        In addition, exception handling is done here.

        """
        try:
            self.finish_request(request, client_address)
            self.shutdown_request(request)
        except:
            self.handle_error(request, client_address)
            self.shutdown_request(request)

    def process_request(self, request, client_address):
        """Start a new thread to process the request."""
        t = threading.Thread(target = self.process_request_thread,
                             args = (request, client_address))
        t.daemon = self.daemon_threads
        t.start()

ThreadingMixIn

图片 36图片 37

class ThreadingTCPServer(ThreadingMixIn, TCPServer): pass

ThreadingTCPServer

RequestHandler相关源码

图片 38图片 39

class BaseRequestHandler:

    """Base class for request handler classes.

    This class is instantiated for each request to be handled.  The
    constructor sets the instance variables request, client_address
    and server, and then calls the handle() method.  To implement a
    specific service, all you need to do is to derive a class which
    defines a handle() method.

    The handle() method can find the request as self.request, the
    client address as self.client_address, and the server (in case it
    needs access to per-server information) as self.server.  Since a
    separate instance is created for each request, the handle() method
    can define arbitrary other instance variariables.

    """

    def __init__(self, request, client_address, server):
        self.request = request
        self.client_address = client_address
        self.server = server
        self.setup()
        try:
            self.handle()
        finally:
            self.finish()

    def setup(self):
        pass

    def handle(self):
        pass

    def finish(self):
        pass

SocketServer.BaseRequestHandler

实例:

图片 40图片 41

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

class MyServer(SocketServer.BaseRequestHandler):

    def handle(self):
        # print self.request,self.client_address,self.server
        conn = self.request
        conn.sendall('欢迎致电 10086,请输入1xxx,0转人工服务.')
        Flag = True
        while Flag:
            data = conn.recv(1024)
            if data == 'exit':
                Flag = False
            elif data == '0':
                conn.sendall('通过可能会被录音.balabala一大推')
            else:
                conn.sendall('请重新输入.')


if __name__ == '__main__':
    server = SocketServer.ThreadingTCPServer(('127.0.0.1',8009),MyServer)
    server.serve_forever()

服务端

图片 42图片 43

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

import socket


ip_port = ('127.0.0.1',8009)
sk = socket.socket()
sk.connect(ip_port)
sk.settimeout(5)

while True:
    data = sk.recv(1024)
    print 'receive:',data
    inp = raw_input('please input:')
    sk.sendall(inp)
    if inp == 'exit':
        break

sk.close()

客户端

源码精简:

图片 44图片 45

import socket
import threading
import select


def process(request, client_address):
    print request,client_address
    conn = request
    conn.sendall('欢迎致电 10086,请输入1xxx,0转人工服务.')
    flag = True
    while flag:
        data = conn.recv(1024)
        if data == 'exit':
            flag = False
        elif data == '0':
            conn.sendall('通过可能会被录音.balabala一大推')
        else:
            conn.sendall('请重新输入.')

sk = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sk.bind(('127.0.0.1',8002))
sk.listen(5)

while True:
    r, w, e = select.select([sk,],[],[],1)
    print 'looping'
    if sk in r:
        print 'get request'
        request, client_address = sk.accept()
        t = threading.Thread(target=process, args=(request, client_address))
        t.daemon = False
        t.start()

sk.close()

View Code

如精简代码能够看出,SocketServer的ThreadingTCPServer之所以能够并且管理供给得益于select 和 Threading 四个东西,其实本质上正是在劳动器端为每多个客商端创设三个线程,当前线程用来拍卖对应客商端的号召,所以,能够支撑相同的时候n个客商端链接(长连接)。

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()时便得到通知。

图片 46

 

功能:

sk =
socket.socket(socket.AF_INET,socket.SOCK_STREAM,0)

参数一:地址簇

  socket.AF_INET
IPv4(默认)
  socket.AF_INET6 IPv6

  socket.AF_UNIX
只好用于单生机勃勃的Unix系统经过间通讯

参数二:类型

  socket.SOCK_STREAM  流式socket ,
for TCP (默认)
  socket.SOCK_DGRAM  
数据报式socket , for UDP

  socket.SOCK_RAW
原始套接字,普通的套接字不可能管理ICMP、I达托霉素P等网络报文,而SOCK_RAW可以;其次,SOCK_RAW也得以管理非常的IPv4报文;其余,利用原始套接字,能够通过IP_HDKugaINCL套接字选项由顾客构造IP头。
  socket.SOCK_HavalDM
是风流罗曼蒂克种保证的UDP情势,即确认保证交到数据报但不保险顺序。SOCK_RAM用来提供对原始公约的初级访问,在急需举行某个特殊操作时使用,如发送ICMP报文。SOCK_RAM平时只限于高端顾客或领队运营的次序行使。
  socket.SOCK_SEQPACKET
可相信的三回九转数据包服务

参数三:协议

  0  (暗中认可)与一定的地址家族有关的磋商,假使是
0 ,则系统就能总局址格式和套接体系,自动接纳一个合适的会谈

图片 47图片 48

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




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()

UDP Demo

sk.bind(address)

  s.bind(address)
将套接字绑定到地点。address地址的格式决意于地址族。在AF_INET下,以元组(host,port)的形式表示地址。

sk.listen(backlog)

  起始监听传入连接。backlog钦点在不肯连接此前,能够挂起的最亚松森接数量。

     
backlog等于5,表示内核已经摄取了连年哀告,但服务器还向来不调用accept实行管理的总是个数最大为5
     
那个值无法Infiniti大,因为要在基础中爱抚连接队列

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()

  套接字的文书陈说符

 

IO 多路复用

I/O多路复用指:通过生机勃勃种体制,可以监视多少个描述符,风流倜傥旦某些描述符就绪(平时是读就绪恐怕写就绪),能够通告顺序开展相应的读写操作。

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中有贰个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 秒,之后返回三个空列表,如果监听的句柄有变化,则直接执行。

图片 49图片 50

#!/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监听终端操作实例

图片 51图片 52

#!/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顾客端哀告:服务端

图片 53图片 54

#!/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版本的服务器端也回天乏术做到何况操作。

图片 55图片 56

#!/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服务端

 

SocketServer 模块

SocketServer内部应用 IO多路复用 以至“八线程” 和 “多过程”
,进而完结产出管理四个客商端乞请的Socket服务端。

+------------+
| BaseServer |
+------------+
      |
      v
+-----------+        +------------------+
| TCPServer |------->| UnixStreamServer |
+-----------+        +------------------+
      |
      v
+-----------+        +--------------------+
| UDPServer |------->| UnixDatagramServer |
+-----------+        +--------------------+

SocketServer简化了网络服务器的编写。它有4个类:TCPServer,UDPServer,UnixStreamServer,UnixDatagramServer。
这4个类是同步进行处理的,另外通过ForkingMixIn和ThreadingMixIn类来支持异步。

ThreadingTCPServer

ThreadingTCPServer完毕的Soket服务器内部会为各种client创造二个“线程”,该线程用来和顾客端实行互动。

1、ThreadingTCPServer基础

使用ThreadingTCPServer:

  • 始建二个无冕自
    SocketServer.BaseRequestHandler 的类
  • 类中必需定义三个称号为 handle
    的方式
  • 启动ThreadingTCPServer

    #!/usr/bin/env python
    # –– coding:utf-8 –
    import SocketServer

    class MyServer(SocketServer.BaseRequestHandler):

    def handle(self):
        pass
    

    if name == ‘main‘:

    server = SocketServer.ThreadingTCPServer(('127.0.0.1',8766), MyServer)
    server.serve_forever()
    

2、ThreadingTCPServer源码解析

ThreadingTCPServer的类图关系如下:

图片 57

其间调用流程为:

  • 开发银行服务端程序
  • 执行 TCPServer.__init__
    方法,创克制务端Socket对象并绑定 IP 和 端口
  • 执行 BaseServer.__init__
    方法,将自定义的接轨自SocketServer.BaseRequestHandler 的类
    MyRequestHandle赋值给self.RequestHandlerClass
  • 执行 BaseServer.server_forever
    方法,While 循环一向监听是不是有顾客端诉求达到 …
  • 当客商端连接达到服务器
  • 执行
    ThreadingMixIn.process_request 方法,创立叁个 “线程”
    用来管理央求
  • 执行 ThreadingMixIn.process_request_thread
    方法
  • 执行 BaseServer.finish_request
    方法,推行 self.RequestHandlerClass()  即:实施 自定义
    MyRequestHandler
    的构造方法(自动调用基类BaseRequestHandler的构造方法,在该构造方法中又会调用
    MyRequestHandler的handle方法)

ThreadingTCPServer相关源码:

图片 58图片 59

class BaseServer:

    """Base class for server classes.

    Methods for the caller:

    - __init__(server_address, RequestHandlerClass)
    - serve_forever(poll_interval=0.5)
    - shutdown()
    - handle_request()  # if you do not use serve_forever()
    - fileno() -> int   # for select()

    Methods that may be overridden:

    - server_bind()
    - server_activate()
    - get_request() -> request, client_address
    - handle_timeout()
    - verify_request(request, client_address)
    - server_close()
    - process_request(request, client_address)
    - shutdown_request(request)
    - close_request(request)
    - handle_error()

    Methods for derived classes:

    - finish_request(request, client_address)

    Class variables that may be overridden by derived classes or
    instances:

    - timeout
    - address_family
    - socket_type
    - allow_reuse_address

    Instance variables:

    - RequestHandlerClass
    - socket

    """

    timeout = None

    def __init__(self, server_address, RequestHandlerClass):
        """Constructor.  May be extended, do not override."""
        self.server_address = server_address
        self.RequestHandlerClass = RequestHandlerClass
        self.__is_shut_down = threading.Event()
        self.__shutdown_request = False

    def server_activate(self):
        """Called by constructor to activate the server.

        May be overridden.

        """
        pass

    def serve_forever(self, poll_interval=0.5):
        """Handle one request at a time until shutdown.

        Polls for shutdown every poll_interval seconds. Ignores
        self.timeout. If you need to do periodic tasks, do them in
        another thread.
        """
        self.__is_shut_down.clear()
        try:
            while not self.__shutdown_request:
                # XXX: Consider using another file descriptor or
                # connecting to the socket to wake this up instead of
                # polling. Polling reduces our responsiveness to a
                # shutdown request and wastes cpu at all other times.
                r, w, e = _eintr_retry(select.select, [self], [], [],
                                       poll_interval)
                if self in r:
                    self._handle_request_noblock()
        finally:
            self.__shutdown_request = False
            self.__is_shut_down.set()

    def shutdown(self):
        """Stops the serve_forever loop.

        Blocks until the loop has finished. This must be called while
        serve_forever() is running in another thread, or it will
        deadlock.
        """
        self.__shutdown_request = True
        self.__is_shut_down.wait()

    # The distinction between handling, getting, processing and
    # finishing a request is fairly arbitrary.  Remember:
    #
    # - handle_request() is the top-level call.  It calls
    #   select, get_request(), verify_request() and process_request()
    # - get_request() is different for stream or datagram sockets
    # - process_request() is the place that may fork a new process
    #   or create a new thread to finish the request
    # - finish_request() instantiates the request handler class;
    #   this constructor will handle the request all by itself

    def handle_request(self):
        """Handle one request, possibly blocking.

        Respects self.timeout.
        """
        # Support people who used socket.settimeout() to escape
        # handle_request before self.timeout was available.
        timeout = self.socket.gettimeout()
        if timeout is None:
            timeout = self.timeout
        elif self.timeout is not None:
            timeout = min(timeout, self.timeout)
        fd_sets = _eintr_retry(select.select, [self], [], [], timeout)
        if not fd_sets[0]:
            self.handle_timeout()
            return
        self._handle_request_noblock()

    def _handle_request_noblock(self):
        """Handle one request, without blocking.

        I assume that select.select has returned that the socket is
        readable before this function was called, so there should be
        no risk of blocking in get_request().
        """
        try:
            request, client_address = self.get_request()
        except socket.error:
            return
        if self.verify_request(request, client_address):
            try:
                self.process_request(request, client_address)
            except:
                self.handle_error(request, client_address)
                self.shutdown_request(request)

    def handle_timeout(self):
        """Called if no new request arrives within self.timeout.

        Overridden by ForkingMixIn.
        """
        pass

    def verify_request(self, request, client_address):
        """Verify the request.  May be overridden.

        Return True if we should proceed with this request.

        """
        return True

    def process_request(self, request, client_address):
        """Call finish_request.

        Overridden by ForkingMixIn and ThreadingMixIn.

        """
        self.finish_request(request, client_address)
        self.shutdown_request(request)

    def server_close(self):
        """Called to clean-up the server.

        May be overridden.

        """
        pass

    def finish_request(self, request, client_address):
        """Finish one request by instantiating RequestHandlerClass."""
        self.RequestHandlerClass(request, client_address, self)

    def shutdown_request(self, request):
        """Called to shutdown and close an individual request."""
        self.close_request(request)

    def close_request(self, request):
        """Called to clean up an individual request."""
        pass

    def handle_error(self, request, client_address):
        """Handle an error gracefully.  May be overridden.

        The default is to print a traceback and continue.

        """
        print '-'*40
        print 'Exception happened during processing of request from',
        print client_address
        import traceback
        traceback.print_exc() # XXX But this goes to stderr!
        print '-'*40

BaseServer

图片 60图片 61

class TCPServer(BaseServer):

    """Base class for various socket-based server classes.

    Defaults to synchronous IP stream (i.e., TCP).

    Methods for the caller:

    - __init__(server_address, RequestHandlerClass, bind_and_activate=True)
    - serve_forever(poll_interval=0.5)
    - shutdown()
    - handle_request()  # if you don't use serve_forever()
    - fileno() -> int   # for select()

    Methods that may be overridden:

    - server_bind()
    - server_activate()
    - get_request() -> request, client_address
    - handle_timeout()
    - verify_request(request, client_address)
    - process_request(request, client_address)
    - shutdown_request(request)
    - close_request(request)
    - handle_error()

    Methods for derived classes:

    - finish_request(request, client_address)

    Class variables that may be overridden by derived classes or
    instances:

    - timeout
    - address_family
    - socket_type
    - request_queue_size (only for stream sockets)
    - allow_reuse_address

    Instance variables:

    - server_address
    - RequestHandlerClass
    - socket

    """

    address_family = socket.AF_INET

    socket_type = socket.SOCK_STREAM

    request_queue_size = 5

    allow_reuse_address = False

    def __init__(self, server_address, RequestHandlerClass, bind_and_activate=True):
        """Constructor.  May be extended, do not override."""
        BaseServer.__init__(self, server_address, RequestHandlerClass)
        self.socket = socket.socket(self.address_family,
                                    self.socket_type)
        if bind_and_activate:
            try:
                self.server_bind()
                self.server_activate()
            except:
                self.server_close()
                raise

    def server_bind(self):
        """Called by constructor to bind the socket.

        May be overridden.

        """
        if self.allow_reuse_address:
            self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.socket.bind(self.server_address)
        self.server_address = self.socket.getsockname()

    def server_activate(self):
        """Called by constructor to activate the server.

        May be overridden.

        """
        self.socket.listen(self.request_queue_size)

    def server_close(self):
        """Called to clean-up the server.

        May be overridden.

        """
        self.socket.close()

    def fileno(self):
        """Return socket file number.

        Interface required by select().

        """
        return self.socket.fileno()

    def get_request(self):
        """Get the request and client address from the socket.

        May be overridden.

        """
        return self.socket.accept()

    def shutdown_request(self, request):
        """Called to shutdown and close an individual request."""
        try:
            #explicitly shutdown.  socket.close() merely releases
            #the socket and waits for GC to perform the actual close.
            request.shutdown(socket.SHUT_WR)
        except socket.error:
            pass #some platforms may raise ENOTCONN here
        self.close_request(request)

    def close_request(self, request):
        """Called to clean up an individual request."""
        request.close()

TCPServer

图片 62图片 63

class ThreadingMixIn:
    """Mix-in class to handle each request in a new thread."""

    # Decides how threads will act upon termination of the
    # main process
    daemon_threads = False

    def process_request_thread(self, request, client_address):
        """Same as in BaseServer but as a thread.

        In addition, exception handling is done here.

        """
        try:
            self.finish_request(request, client_address)
            self.shutdown_request(request)
        except:
            self.handle_error(request, client_address)
            self.shutdown_request(request)

    def process_request(self, request, client_address):
        """Start a new thread to process the request."""
        t = threading.Thread(target = self.process_request_thread,
                             args = (request, client_address))
        t.daemon = self.daemon_threads
        t.start()

ThreadingMixIn

图片 64图片 65

class ThreadingTCPServer(ThreadingMixIn, TCPServer): pass

ThreadingTCPServer

RequestHandler相关源码

图片 66图片 67

class BaseRequestHandler:

    """Base class for request handler classes.

    This class is instantiated for each request to be handled.  The
    constructor sets the instance variables request, client_address
    and server, and then calls the handle() method.  To implement a
    specific service, all you need to do is to derive a class which
    defines a handle() method.

    The handle() method can find the request as self.request, the
    client address as self.client_address, and the server (in case it
    needs access to per-server information) as self.server.  Since a
    separate instance is created for each request, the handle() method
    can define arbitrary other instance variariables.

    """

    def __init__(self, request, client_address, server):
        self.request = request
        self.client_address = client_address
        self.server = server
        self.setup()
        try:
            self.handle()
        finally:
            self.finish()

    def setup(self):
        pass

    def handle(self):
        pass

    def finish(self):
        pass

SocketServer.BaseRequestHandler

SocketServer的ThreadingTCPServer之所以能够同期管理央求得益于 select 和 Threading 多少个东西,其实本质上就是在劳务器端为每二个客商端成立三个线程,当前线程用来拍卖对应客商端的乞请,所以,可以支撑同一时间n个客商端链接(长连接)。

ForkingTCPServer

ForkingTCPServer和ThreadingTCPServer的使用和施行流程基本豆蔻梢头致,只然则在里边分别为央浼者创设“线程”  和 “进度”。

主导采取:

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

class MyServer(SocketServer.BaseRequestHandler):

    def handle(self):
        pass

if __name__ == '__main__':
    server = SocketServer.ForkingTCPServer(('127.0.0.1',8009),MyServer)
    server.serve_forever()