1、socket语法及相关
在网络编程中个一个基本组件就是套接字(socket)。套接字基本上是两个端点的程序之间的'信息通道'。程序可能分布在不同的计算机上(通过网络连接),通过套接字相互发送信息。在Python中大多数的网络编程都隐藏了socket模块的基本细节,不直接和套接字交互。
套接字包括两个:服务器套接字和客户机套接字。在创建一个服务器套接字后,让它等待连接。这样它就在某个网络地址处(IP地址和一个端口号的组合)监听,直到有客户机套接字连接。连接完成后,两者就可以进行交互了。
socket起源于Unix,而Unix/Linux基本哲学之一就是“一切皆文件”,对于文件用【打开】【读写】【关闭】模式来操作。socket就是该模式的一个实现,socket即是一种特殊的文件,一些socket函数就是对其进行的操作(读/写IO、打开、关闭)
socket和file的区别:
- file模块是针对某个指定文件进行【打开】【读写】【关闭】
- socket模块是针对 服务器端 和 客户端Socket 进行【打开】【读写】【关闭】
import socket server=socket.socket() #创建服务端套接字 server.bind(('localhost',8888))#绑定地址 server.listen(5)#监听连接 print("我要开始等电话了") while True: conn,addr=server.accept()#接收请求,等待电话打进来 #conn 就是客户端打进来而在其服务端为其生成一个连接实例 print(conn, addr) print("电话来了") while True: data=conn.recv(1024)#接收信息 print('recv:', data) print('recv:', data.decode()) if not data: print('client has lost') break conn.send(data.upper()) #conn.close() 关闭客户端套接字 continue server.close()
import socket client=socket.socket()#创建客户端套接字 client.connect(('localhost',8888))#尝试连接服务器 # client.send(b'Hello World') while True: msg=input(">>:").strip() if len(msg)==0: #发不了空值,发空值会卡住 continue client.send(msg.encode('utf-8')) data=client.recv(1024) print("recv:",data.decode()) client.close()#关闭客户端套接字
#通过socket实现减半的ssh import socket,os server=socket.socket() server.bind(('localhost',9999)) server.listen() while True: conn,addr=server.accept() print("new conn:",addr) while True: data=conn.recv(1024)#接收少于1024k的数据 if not data: print('客户机已断开') break print('执行指令:',data) cmd_res=os.popen(data.decode()).read()#接收字符串,执行结果也是字符串 print('before send:',len(cmd_res)) if len(cmd_res)==0: cmd_res='cmd has no output' conn.send(str(len(cmd_res.encode('utf-8'))).encode('utf-8')) conn.send(cmd_res.encode('utf-8')) print('send done') server.close()
import socket client=socket.socket() client.connect(('localhost',9999)) while True: cmd=input('>>:').strip() if len(cmd)==0:continue client.send(cmd.encode('utf-8'))#需要转成字节类型 data_size=client.recv(1024) receive_size=0 print(data_size) while receive_size<int(data_size): #直到完成接收完整才结束 data=client.recv(1024) receive_size+=int(len(data))#每次接收的有可能少于1024k,所以必须要用len判断 print(receive_size) print(data.decode()) client.close()
#通过socket实现减半的ssh import socket,os server=socket.socket() server.bind(('localhost',9999)) server.listen() while True: conn,addr=server.accept() print("new conn:",addr) while True: data=conn.recv(1024)#接收少于1024k的数据 if not data: print('客户机已断开') break print('执行指令:',data) cmd_res=os.popen(data.decode()).read()#接收字符串,执行结果也是字符串 print('before send:',len(cmd_res)) if len(cmd_res)==0: cmd_res='cmd has no output' # conn.send(str(len(cmd_res.encode('utf-8'))).encode('utf-8')) # conn.send(cmd_res.encode('utf-8')) #由于这两个send紧挨着,有时可能出现粘包问题,因为缓冲区可能会把这两条数据合并成一条发送给客户端 #解决办法 conn.send(str(len(cmd_res.encode('utf-8'))).encode('utf-8')) client_ack=conn.recv(1024) #wait client to confirm print('the client confirm:',client_ack.decode()) conn.send(cmd_res.encode('utf-8')) print('send done') server.close()
import socket client=socket.socket() client.connect(('localhost',9999)) while True: cmd=input('>>:').strip() if len(cmd)==0:continue client.send(cmd.encode('utf-8'))#需要转成字节类型 data_size=client.recv(1024) client.send('准备好接收了'.encode('utf-8'))#接收提示 receive_size=0 print(data_size) while receive_size<int(data_size): #直到完成接收完整才结束 data=client.recv(1024) receive_size+=int(len(data))#每次接收的有可能少于1024k,所以必须要用len判断 print(receive_size) print(data.decode()) client.close()
WEB服务应用:
import socket def handle_request(client): buf = client.recv(1024) client.send("HTTP/1.1 200 OK ") 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 IPv6socket.AF_UNIX 只能够用于单一的Unix系统进程间通信
参数二:类型
socket.SOCK_STREAM 流式socket , for TCP (默认)
socket.SOCK_DGRAM 数据报式socket , for UDPsocket.SOCK_RAW 原始套接字,普通的套接字无法处理ICMP、IGMP等网络报文,而SOCK_RAW可以;其次,SOCK_RAW也可以处理特殊的IPv4报文;此外,利用原始套接字,可以通过IP_HDRINCL套接字选项由用户构造IP头。
socket.SOCK_RDM 是一种可靠的UDP形式,即保证交付数据报但不保证顺序。SOCK_RAM用来提供对原始协议的低级访问,在需要执行某些特殊操作时使用,如发送ICMP报文。SOCK_RAM通常仅限于高级用户或管理员运行的程序使用。
socket.SOCK_SEQPACKET 可靠的连续数据包服务参数三:协议
0 (默认)与特定的地址家族相关的协议,如果是 0 ,则系统就会根据地址格式和套接类别,自动选择一个合适的协议
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()
服务器套接字方法 | |
s.bind() | 将地址(主机名、端口号对)邦定到套接字上 |
s.listen() | 设置并启动TCP监听器 |
s.accept() | 被动接受TCP客户端连接,一直等待连接到达(阻塞) |
客户端套接字方法 | |
s.connect() | 主动发起TCP服务器连接 |
s.connect_ex() | connect()的扩展版本,此时会一以错误码的形式返回问题,而不是抛出一个异常 |
普通套接字方法 | |
s.recv() | 接收TCP消息 |
s.recv_into | 接收TCP消息到指定的缓冲区 |
s.send() | 发送TCP消息 |
s.sendall() | 完整地发送TCP消息 |
s.recvfrom() | 接收UDP消息 |
s.recvfrom_into() | 接收UDP消息到指定的缓冲区 |
s.sendto() | 发送UDP消息 |
s.getpeername() | l连接到套接字(TCP)的远程地址 |
s.getsocketname() | 当前套接字地址 |
s.getsockopt() | 返回给定的套接字选项的值 |
s.setsockopt() | 设置给定套接字的值 |
s.shutdown() | 关闭连接 |
s.close() | 关闭套接字 |
s.detach() | 在未关闭文件描述符的情况下关闭套接字,返回文件描述符 |
s.ioctl() | 控制套接字的模式(仅支持Windows) |
面向阻塞的套接字方法 | |
s.setblocking() | 设置套接字的阻塞或费阻塞模式 |
s.settimeout() | 设置阻塞套接字操作的超时时间 |
s.gettimeout() | 获取阻塞套接字的操作的超时时间 |
面向文件的套接字方法 | |
s.fileno() | 套接字的文件描述符 |
s.makefile() | 创建与套接字关联的文件对象 |
数据属性 | |
s.family() | 套接字家族 |
s.type() | 套接字类型 |
s.proto() | 套接字协议 |
部分常用套接字详解:
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()
套接字的文件描述符
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() 套接字的文件描述符
socket模块属性
除了socket.socket()函数之外,socket模块还提供了更多用于网络应用开发的属性,下面列出一些比较受欢迎的属性
属性名称 | 描述 |
数据属性 | |
AF_UNIX、AF_INET、AF_INET6 AF_NETLINK、AF_TIPC |
Python中支持的套接字地址家族 |
SO_STREAM、SO_DGRAM |
套接字类型(TCP=流,UDP=数据报) |
has_ipv6 | 指示是否支持IPv6的布尔标记 |
异常 | |
error | 套接字相关错误 |
heror | 主机和地址相关错误 |
gaierror | 地址相关错误 |
timeout | 超时时间 |
函数 | |
socket() | 以给定的地址家族、套接字类型和协议类型(可选)创建一个套接字对象 |
socketpair() | 以给定的地址家族、套接字类型和协议类型(可选)创建一对套接字对象 |
create_connection() | 常规函数,它接收一个地址(主机名,端口号)对,返回套接字对象 |
fromfd() | 以一个打开的文件描述符创建一个套接字对象 |
ssl() | 通过套接字启动一个安全套接字层连接;不执行证书验证 |
getaddrinfo() |
获取一个五元组序列形式的地址信息 |
getnameinfo() | 给定一个套接字地址,返回(主机名,端口号)二元组 |
getfqdn() | 返回完整的域名 |
getostname() | 返回当前主机名 |
getostbyname() | 将一个主机名映射到它的IP地址 |
getostbyname_ex() | getostbyname()的扩展版本,它返回主机名、别主机集合和IP地址列表 |
getostbyaddr() | 将一个IP地址映射到DNS信息,返回与getostbyname_ex()相同的3元组 |
getprotobyname() | 将一个协议名(如‘tcp’)映射到一个数字 |
getservbyname()/ getservbyport() |
将一个服务名映射到一个端口号,或者反过来;对于任何一个函数来说,协议名多事可选的 |
ntohl()/ntohs() | 将来自2网络的整数转换为主机字节顺序 |
htonl()/htons() | 将来自主机的整数转换为网络字节顺序 |
inet_aton()/inet_ntoa() | 将IP地址八进制字符串转换为32位的包格式,或者反过来(仅用于IPv4地址) |
inet_pton()/inet+ntop() |
将iP地址字符串转换成打包的二进制格式,或者反过来(同时适用于IPv4和IPv6地址) |
getdefaulttimeout()/setdefaulttimeout() | 以秒(浮点数)为单位返回默认套接字超时时间,以秒(浮点数)为单位设置默认套接字超时时间 |
2、SocketServer模块
SocketServer模块是标准库中很多服务器框架的基础。SocketServer简化了网络服务器的编写。它有4个类:针对TCP流式套接字的TCPServer,针对UDP数据报套接字的UDPServer,以及针对性不强的UnixStreamServer和UnixDatagramServer。我们能用到的一般是TCPServer,后面三个很少用得到,这4个类是同步进行处理的,另外通过ForkingMixIn和ThreadingMixIn类来支持异步。
SocketServer的实现原理
如果要编写一个使用SocketServer框架的服务器,可能会将大部分代码放在一个请求处理程序(request handler)中。每当服务器收到一个请求(来客户端的连接)时,就会实例化一个请求处理程序,并且它的各种处理方法(handler method)会在处理请求时被调用。具体调用哪个方法取决于特定的服务器和使用的处理程序类(handler class)。还可以把它们子类化,使得服务器调用自定义的处理程序集。BaseRequestHandler类把所有的惭怍都放到了处理器的一个键handle的方法中,这个方法会被服务器调用,然后这个方法就会访问self.request中的客户端套接字。
SocketServer创建一个服务的步骤:
1.创建一个request handler class(请求处理类),继承自BaseRequestHandler class并重写它的handle()方法,该方法将处理到的请求。
2.实例化一个server class对象,并将服务的地址和之前创建的request handler class传递给它。
3.调用server class对象的handle_request() 或 serve_forever()方法来开始处理请求。
from SocketServer import TCPServer,StreamRequestHandler #定义请求处理类 class Handler(StreamRequestHandler): def handle(self): addr = self.request.getpeername() print( 'Got connection from ',addr ) self.wfile.write('Thank you for connecting') server = TCPServer(('',1234), Handler) #实例化服务类对象 server.server_forever() #开启服务
一个基于SocketServer的小型接收文件服务器:
import socket import sys import os ip_port=('127.0.0.1',9999) sk=socket.socket() sk.connect(ip_port) container={'key':'','data':''} while True: input=input('path:') cmd,path=input.split('|') file_name=os.path.basename(path) file_size=os.stat(path).st_size sk.send(cmd+"|"+file_name+'|'+str(file_size)) send_size=0 f=file(path,'rb') Flag=True while Flag: if send_size+1024>file_size: data=f.read(file_size-send_size) Flag=False else: data=f.read(1024) send_size+=1024 sk.send(data) f.flose() sk.close()
import SocketServer import os class MyServer(SocketServer.BaseRequestHandler): def handle(self): base_path='G:/temp' conn=self.request print('connected...') while True: pre_data=conn.recv(1024) #获取请求方法、文件名、文件大小 cmd,file_name,file_size=pre_data.split('|') #已经接收文件的大小 recv_size=0 #上传文件路径 file_dir=os.path.join(base_path,file_name) f=file(file_dir,'wb') Flag=True while Flag: #未上传完毕, if int(file_size)>recv_size: # 最多接收1024,可能小于1024 data=conn.recv(1024) recv_size+=len(data) #上传完毕 else: recv_size=0 Flag=False #写入文件 f.write(data) print('upload succssed') f.close() instance=SocketServer.ThreadingTCPServer(('127.0.0.1',9999),MyServer) instance.serve_forever()
SocketServer的异步实现
SocketServer内部使用 IO多路复用 以及 “多线程” 和 “多进程” ,从而实现并发处理多个客户端请求的Socket服务端。即:每个客户端请求连接到服务器时,Socket服务端都会在服务器是创建一个“线程”或者“进程” 专门负责处理当前客户端的所有请求。
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): # 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()
#!/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的类图关系如下:
内部调用流程为:
- 启动服务端程序
- 执行 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相关源码:
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
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()
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()
class ThreadingTCPServer(ThreadingMixIn, TCPServer): pass
RequestHandler相关源码:
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
实例:
#!/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()
#!/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()
源码精简:
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()
如精简代码可以看出,SocketServer的ThreadingTCPServer之所以可以同时处理请求得益于 select 和 Threading 两个东西,其实本质上就是在服务器端为每一个客户端创建一个线程,当前线程用来处理对应客户端的请求,所以,可以支持同时n个客户端链接(长连接)。
ForkingTCPServer
ForkingTCPServer和ThreadingTCPServer的使用和执行流程基本一致,只不过在内部分别为请求者建立 “线程” 和 “进程”。
基本使用:
#!/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.ForkingTCPServer(('127.0.0.1',8009),MyServer) server.serve_forever()
#!/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()
以上ForkingTCPServer只是将 ThreadingTCPServer 实例中的代码:
server=SocketServer.ThreadingTCPServer(('127.0.0.1',8009),MyRequestHandler) 变更为: server=SocketServer.ForkingTCPServer(('127.0.0.1',8009),MyRequestHandler)
SocketServer的ThreadingTCPServer之所以可以同时处理请求得益于 select 和 os.fork 两个东西,其实本质上就是在服务器端为每一个客户端创建一个进程,当前新创建的进程用来处理对应客户端的请求,所以,可以支持同时n个客户端链接(长连接)。
源码剖析参考 ThreadingTCPServer
参考自:http://www.cnblogs.com/wupeiqi/articles/5040823.html