标签:参数 数据包 start log mat eth should 匹配 面向连接
比如客户端与服务端的通信,是需要跨越应用层、传输层、网络层以及链路层的,应用层也就是我们用户接触到的层(主要就是HTTP协议所在的层),包括一些应用程序;传输层主要是TCP/UDP协议所在的层,其作用就是传输数据包;网络层主要是IP协议所在的层,其作用就是传输数据包寻找一条合适的路径;最后是链路层,主要是以太网协议所在的层,它与硬件打交道。
传输层(TCP/UDP协议)以及网络层(IP协议)都是非常复杂的,如果直接与其打交道,是非常困难的,此时socket就上场了,那么Socket是什么呢?
socket也被叫做套接字,它就是在应用层与传输层之间的一个抽象层,应用程序可以通过它发送或接收数据,可对其进行像对文件一样的打开、读写和关闭等操作。
两个应用程序进行通讯时,一个应用程序会将其发送的信息写入到它所在主机的socket中,该socket通过网络接口卡的传输介质将这段信息发送给另一台主机的socket中,使这段信息能传送到其他程序中。因此,两个应用程序之间的数据传输要通过套接字(socket)来完成。
每个套接字(socket)都有一个序号,这个序号是由IP地址和端口号组成的,所以套接字Socket=(IP地址:端口号)。
在网络应用程序设计时,由于TCP/IP的核心内容被封装在操作系统中,如果应用程序要使用TCP/IP,可以通过系统提供的TCP/IP的编程接口来实现。因此也就有基于TCP的套接字与基于UDP的套接字。
服务端先初始化socket然后与端口进行绑定(通过bind方法),监听端口(通过listen方法),调用accept方法进行阻塞直到有客户端来连接它。
客户端也会初始化一个socket,然后通过connect方法来连接服务端监听的端口,当connectz执行成功后它们之间的连接就建立了。
此时如果客户端发送消息,服务端接收请求数据并进行处理,最后把回应数据返回给客户端,最后关闭连接。这样完成一次交互。
sock_ser= socket.socket(socket.AF_INET,socket.SOCK_STREAM)
sock_ser.bind((‘127.0.0.1‘,8000))
sock_ser.listen()
conn,addr = sock_ser.accept()
msg = conn.recv(1024) #收消息 conn.send(msg.upper())#发消息
conn.close()
sock.close()
上面是服务端的Socket创建过程,客户端不需要绑定ip和端口,只是使用connect方法连接服务端监听的ip和端口。
服务端sock_ser.py
import socket #创建socket对象 sock_ser = socket.socket(socket.AF_INET,socket.SOCK_STREAM) #绑定ip和端口 sock_ser.bind((‘127.0.0.1‘,8000)) #监听ip和端口 sock_ser.listen(5) #阻塞方法,等待客户端连接 conn,addr = sock_ser.accept() #接收客户端发送的消息 msg = conn.recv(1024) #收消息 print(‘客户端发来的消息是:‘, msg) #客户端发来的消息是: b‘\xe4\xbd\xa0\xe5\xa5\xbd‘ #发送消息 conn.send(msg.upper()) #关闭连接 conn.close() #关闭套接字socket sock_ser.close()
客户端sock_cli.py
import socket #创建socket对象 sock_cli = socket.socket(socket.AF_INET,socket.SOCK_STREAM) #连接服务端监听的ip和端口 sock_cli.connect(("127.0.0.1",8000)) #发送消息 sock_cli.send(bytes("你好",encoding="utf-8")) #接收消息 data = sock_cli.recv(1024) print(‘服务端发来的消息‘,data) #服务端发来的消息 b‘\xe4\xbd\xa0\xe5\xa5\xbd‘ #关闭套接字socket对象 sock_cli.close()
注意:收发消息是字节形式,所以需要将字符串转成字节。
上面显然完成的是一次交互,服务端与客户端这样收发消息后程序就结束了,那么如何能够让服务端与多个客户端持续交互呢?
import socket #创建socket对象 sock_ser = socket.socket(socket.AF_INET,socket.SOCK_STREAM) #绑定ip和端口 sock_ser.bind((‘127.0.0.1‘,8000)) #监听ip和端口 sock_ser.listen(5) #不断循环连接,等待多个客户端连接 while True: conn, addr = sock_ser.accept() #不断循环通信,与客户端连续交互 while True: msg = conn.recv(1024) #收消息 if msg: print(‘客户端发来的消息是:‘, msg.decode()) #客户端发来的消息是: b‘\xe4\xbd\xa0\xe5\xa5\xbd‘ # 发送消息 conn.send(msg.upper()) else: break #如果没有消息发送过来就退出循环 #关闭连接 conn.close() #关闭套接字socket sock_ser.close()
# import socket # # #创建socket对象 # sock_cli = socket.socket(socket.AF_INET,socket.SOCK_STREAM) # # #连接服务端监听的ip和端口 # sock_cli.connect(("127.0.0.1",8000)) # # #发送消息 # sock_cli.send(bytes("你好",encoding="utf-8")) # # #接收消息 # data = sock_cli.recv(1024) # print(‘服务端发来的消息‘,data) #服务端发来的消息 b‘\xe4\xbd\xa0\xe5\xa5\xbd‘ # # #关闭套接字socket对象 # sock_cli.close() import socket #创建socket对象 sock_cli = socket.socket(socket.AF_INET,socket.SOCK_STREAM) #连接服务端监听的ip和端口 sock_cli.connect(("127.0.0.1",8000)) while True: #发送消息 msg = input(‘>>:‘) if msg: sock_cli.send(msg.encode()) #将字符串转成字节,或者sock_cli.send(bytes(msg,encoding=""utf-8)) #接收消息 data = sock_cli.recv(1024) print(‘服务端发来的消息‘,data.decode()) #服务端发来的消息 b‘\xe4\xbd\xa0\xe5\xa5\xbd‘ else: break #关闭套接字socket对象 sock_cli.close()
上述虽然可以实现与多个客户端持续交互,但是出现的问题就是如果第一个客户端保持连接,后面的客户端连接上了但会卡住?
这是因为服务端还陷入与第一个客户端交互的循环中,如果断开第一个客户端,服务端的conn也会断掉,服务端抛出异常,为了保证断开第一个客户端而持续的处理后面客户端的请求,服务端可以加入异常处理:
import socket #创建socket对象 sock_ser = socket.socket(socket.AF_INET,socket.SOCK_STREAM) #绑定ip和端口 sock_ser.bind((‘127.0.0.1‘,8000)) #监听ip和端口 sock_ser.listen(5) # 不断循环连接,等待多个客户端连接 while True: conn, addr = sock_ser.accept() #不断循环通信,与客户端连续交互 while True: #加入异常处理,如果有客户端断开,服务端不会报错,继续处理下一个连接的请求 try: msg = conn.recv(1024) #收消息 if msg: print(‘客户端发来的消息是:‘, msg.decode()) #客户端发来的消息是: b‘\xe4\xbd\xa0\xe5\xa5\xbd‘ # 发送消息 conn.send(msg.upper()) else: break #如果没有消息发送过来就退出循环 except Exception as e: break #关闭连接 conn.close() #关闭套接字socket sock_ser.close()
(1)套接字家族的名字:AF_UNIX
unix一切皆文件,基于文件的套接字调用的就是底层的文件系统来取数据,两个套接字进程运行在同一机器,可以通过访问同一个文件系统间接完成通信
(2)套接字家族的名字:AF_INET
还有AF_INET6被用于ipv6,还有一些其他的地址家族,不过,他们要么是只用于某个平台,要么就是已经被废弃,或者是很少被使用,或者是根本没有实现,所有地址家族中,
AF_INET是使用最广泛的一个,python支持很多种地址家族,但是由于我们只关心网络编程,所以大部分时候只使用AF_INET
(1)类型名称:流式套接字(SOCK_STREAM) 传输层基于tcp协议 的套接字编程方案。 (2)类型名称:数据报套接字(SOCK_DGRAM) 传输层基于udp协议的套接字编程方案。 (3)底层套接字(SOCK_RAM)
访问底层协议的套接字编程。 # 面向连接的传输--tcp协议--可靠地--流式套接字 #面向无连接传输--udp协议--不可靠--数据报套接字
将套接字绑定到address,套接字必须尚未绑定。说明address指的是元祖("127.0.0.0",80)
backlog用于指定队列的长度,等待处理的进入连接的个数最多不能超过这个数字,否则往后的连接将被拒绝,导致客户的连接请求失败。调用后,程序一直会监听这个IP端口,如果有连接请求,就把它加入到这个队列中。
接受连接。套接字必须绑定到一个地址并监听连接。返回值是一对,其中conn是可用于在连接上发送和接收数据的新套接字对象,而 address是绑定到连接另一端上的套接字的地址,(conn,address)
。
将数据发送到套接字。套接字必须连接到远程套接字。返回发送的字节数。
将数据发送到套接字。套接字必须连接到远程套接字。与send
方法不同,此方法继续从字节发送数据,直到发送完所有数据或发生错误为止。 None
成功返回。如果出错,则会引发异常,并且无法确定成功发送了多少数据(如果有)。
从套接字接收数据。返回值是一个字节对象,代表接收到的数据。一次要接收的最大数据量由bufsize指定。
注意:为了与硬件和网络实际情况进行最佳匹配,bufsize的值 应为2的相对较小的幂,例如4096。
从套接字接收数据。返回值是一对 ,其中bytes是表示接收到的数据的字节对象,而address是发送数据的套接字的地址。
连接到的地址为远程套接字,该方法将等待直到连接完成。
关闭连接或套接字对象,套接字在被垃圾回收时会自动关闭,但建议显式调用close关闭套接字。
TCP套接字的特点:TCP套接字是基于连接的,因此在启动时应该先启动服务端,再启动客户端。
在上面实例中已经演示了TCP的套接字
UDP套接字的特点:UDP是无链接的,先启动哪一端都不会报错,并且可以同时与多个客户端通信
服务端 udp_ser.py
import socket #创建sock对象 udp_ser = socket.socket(socket.AF_INET,socket.SOCK_DGRAM) #绑定主机、端口 udp_ser.bind((‘127.0.0.1‘,8000)) #通信循环,可以不断的交互 while True: #接收消息,返回的是接收数据和对面地址的元组 data,addr = udp_ser.recvfrom(1024) print(data.decode()) #发送消息 udp_ser.sendto("response msg...".encode(),addr) #关闭socket对象 udp_ser.close()
客户端 udp_cli.py
import socket #创建sock对象 udp_cli = socket.socket(socket.AF_INET,socket.SOCK_DGRAM) #增加通信循环 while True: msg = input(">>:") if msg: #发送数据 udp_cli.sendto(msg.encode(),(‘127.0.0.1‘,8000)) #接收数据 data,addr = udp_cli.recvfrom(1024) print(‘客户端接收的消息:‘,data.decode()) else: break #关闭socket对象 udp_cli.close()
(1)TCP套接字使用字节流的方式传输,UDP套接字使用数据报形式传输数据 (2)TCP套接字会有粘包现象,UDP套接字有消息边界不会形成粘包 (3)TCP套接字可以保障数据传输完整性,UDP套接字则不保证 (4)TCP套接字需要进行listen accept操作,UDP套接字不需要 (5)TCP套接字收发消息使用新的套接字recv send。UDP套接字使用recvfrom,sendto
注意:总结中的第5条,TCP套接字收发消息使用新的套接字recv send,这个新的套接字是accept()接受一个客户端的连接请求,并返回一个新的套接字(与原套接字不是同一个)。这个“新的套接字”负责与本次接受的客户端的通信(包括收发消息)。
TCP套接字接收消息会发生粘包问题。因为TCP套接字是基于TCP协议的,而TCP协议传输信息是以流的形式,接收端为了提高效率就会将数据量小的包合并成一个大包,这样就很难判断出包中数据的边界。但是这种协议保证了不会丢掉数据,因为如果规定每次接收1024大小,下次还会接着再从此处开始接收。
UDP套接字接收消息不会发生粘包问题。因为它是基于UDP协议的,而UDP协议的传输信息是以消息的形式,所谓的消息可以认为对方一次性write/send的数据为一个消息。UDP的recvfrom是阻塞的,一个recvfrom(x)必须对唯一一个sendinto(y),收完了x个字节的数据就算完成,若是y>x数据就丢失,这意味着UDP根本不会粘包,但是会丢数据,不可靠。
粘包的问题就是接收端不知道发送端发送的字节流长度,如果在发送整个字节流之前先将字节流的大小发送出去,然后接收端来一个死循环接收完所有数据就可以了。
我们在给客户端发送具体的数据之前,先将内容的长度计算出来,然后使用struct模块将任意长度(无论内容长度是100还是1000)打包成4个长度,这样就有一个固定的长度了,服务端先将内容长度发送过去,然后将具体内容再发送过去。
在客户端,我们使用recv方法接收4个长度,这是内容长度打包后的整型长度,然后解包,取出真实内容的长度大小,使用while循环得到所有的数据。
服务端 sock_ser.py
import socket import struct #创建socket对象 sock_ser = socket.socket(socket.AF_INET,socket.SOCK_STREAM) #绑定ip和端口 sock_ser.bind((‘127.0.0.1‘,8000)) #监听ip和端口 sock_ser.listen(5) # 不断循环连接,等待多个客户端连接 while True: conn, addr = sock_ser.accept() #不断循环通信,与客户端连续交互 while True: #加入异常处理,如果有客户端断开,服务端不会报错,继续处理下一个连接的请求 try: msg = conn.recv(1024) #收消息 if msg: print(‘客户端发来的消息是:‘, msg.decode()) #客户端发来的消息是: b‘\xe4\xbd\xa0\xe5\xa5\xbd‘ #解决粘包 #将发送的内容的任意长度打包成4个长度,并发送 msg_length = struct.pack(‘i‘,len(msg)) conn.send(msg_length) # 发送消息 conn.send(msg.upper()) else: break #如果没有消息发送过来就退出循环 except Exception as e: break #关闭连接 conn.close() #关闭套接字socket sock_ser.close()
客户端 sock_cli.py
import socket import struct #创建socket对象 sock_cli = socket.socket(socket.AF_INET,socket.SOCK_STREAM) #连接服务端监听的ip和端口 sock_cli.connect(("127.0.0.1",8000)) while True: msg = input(‘>>:‘) if msg: #发送消息 sock_cli.send(msg.encode()) #将字符串转成字节,或者sock_cli.send(bytes(msg,encoding=""utf-8)) #取出固定长度为4的内容长度打包后的数据 data_header = sock_cli.recv(4) #用struct反解,接收到内容长度,以元组形式呈现,取出第一个值,直接就是整型 data_header_length = struct.unpack(‘i‘,data_header)[0] print("接收消息的长度",data_header) #循环接收消息 recv_data = b‘‘ recv_size = 0 while recv_size < data_header_length: recv_data += sock_cli.recv(recv_size) recv_size = len(recv_data) print(‘服务端发来的消息‘,recv_data.decode()) #服务端发来的消息 b‘\xe4\xbd\xa0\xe5\xa5\xbd‘ else: break #关闭套接字socket对象 sock_cli.close()
>>> import struct >>> struct.pack(‘i‘,100) b‘d\x00\x00\x00‘ >>> msg_length = struct.pack(‘i‘,100) >>> len(msg_length) 4 >>> msg_length = struct.pack(‘i‘,500) >>> len(msg_length) 4 >>> struct.unpack(‘i‘,msg_length) (500,) #反解出来的是一个元祖,所以客户端反解出来后需要取出元祖的第一个值 >>>
在上述socket中的服务端中实现了两层循环。外层循环是连接循环,也就是可以接受多个客户端连接;内层循环是通信循环,也就是可以与客户端不间断的交互,但是服务端只能与一个客户端交互,其它的客户端虽然连接上了但需要等待,因为服务端一直呆在那一个客户端的内层通信循环内。
那么如何解决这个问题呢?使用socketserver模块,它将socket模块与多线程或多进程技术进行结合,从而服务端可以处理并发情况。
import socketserver class MyServer(socketserver.BaseRequestHandler): def handle(self): #通信循环 while True: # 使用异常处理,防止某个客户端断开连接报错 try: # 收消息 data = self.request.recv(1024) if not data: break print(‘收到客户端的消息是‘, data) # 发消息 self.request.sendall(data.upper()) except Exception as e: break if __name__ == ‘__main__‘: s = socketserver.ThreadingTCPServer(("127.0.0.1",8080),MyServer) #多线程实现并发 s.serve_forever() #相当于连接循环
from socket import * tcp_client=socket(AF_INET,SOCK_STREAM) tcp_client.connect((‘127.0.0.1‘,8080)) while True: msg=input(‘>>: ‘).strip() if not msg:break tcp_client.send(msg.encode()) data=tcp_client.recv(1024) print(‘收到服务端发来的消息:‘,data.decode()) tcp_client.close()
客户端和以前还是一样,但是可以多个客户端进行连接同时被响应。
socketserver模块中的内容分为两类,一类是解决连接循环(和Server相关),另一类是解决通信循环(和Request相关)。
在上述的tcp_server.py中,执行:
s=socketserver.ThreadingTCPServer(("127.0.0.1",8080),MyServer)
class ThreadingTCPServer(ThreadingMixIn, TCPServer): pass
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.socket.bind(self.server_address) 绑定主机和端口 self.server_activate() #elf.socket.listen(self.request_queue_size) 监听 except: self.server_close() raise
ThreadingTCPServer中传入的参数会在TCPServer的__init__方法中初始化,此时实例化对象s(也就是self)会获取server_address、RequestHandlerClass(通过基类BaseServer初始化实现)属性以及实例s获取套接字对象socket,并且会绑定主机和端口,监听。
接下来就是执行:
s.serve_forever()
寻找server_forever方法还是先从实例类本身开始,然后是从左向右的继承类。
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: # 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. with _ServerSelector() as selector: selector.register(self, selectors.EVENT_READ) while not self.__shutdown_request: ready = selector.select(poll_interval) if ready: self._handle_request_noblock() #查看这个方法 self.service_actions() finally: self.__shutdown_request = False self.__is_shut_down.set()
def _handle_request_noblock(self): """Handle one request, without blocking. I assume that selector.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() #self.socket.accept()所以request相当于连接conn except OSError: 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) else: self.shutdown_request(request)
在上面的代码中执行self.get_request方法,实际执行的就是self.socket.accept,它返回的就是conn以及addr,只不过在这里用request来代替连接conn。另外它还执行了self.process_request(request, client_address),这是和并发相关,接下来看看到底里面做了什么。
寻找方法时注意一定要按照从本身然后继承的类顺序寻找。
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()
在process_request方法中它接受的参数就是request以及客户端的地址(client_address),每来一个客户端就会建立一个request,然后启动一个线程去处理这个请求request,那么请求具体是怎么处理呢?
上述ThreadingMixIn类的process_request_thread方法中调用了BaseServer类的finish_request:
def finish_request(self, request, client_address): """Finish one request by instantiating RequestHandlerClass.""" self.RequestHandlerClass(request, client_address, self)#RequestHandlerClass就是自己定义的MyServer类
RequestHandlerClass类在初始化时已经赋值给了实例s,所以可以直接通过self.RequestHandlerClass进行初始化,也就是MyServer类的初始化。
MyServer类没有__init__方法,所以调用父类BaseRequestHandler的初始化方法:
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 other arbitrary instance variables. """ def __init__(self, request, client_address, server): self.request = request self.client_address = client_address self.server = server self.setup() try: self.handle() #只要实例化就默认执行handle方法,通信循环 finally: self.finish() def setup(self): pass def handle(self): #自己定义处理每一个线程(请求)的方法 pass def finish(self): pass
socketserver模块中与客户端连接和通信本质还是socket模块,每次连接过来后会起一个线程去处理实现并发,当然socketserver模块还有多进程(ForkingTCPServer)实现并发
class ThreadingTCPServer(ThreadingMixIn, TCPServer): pass
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 selector 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.socket.bind(self.server_address) 绑定主机和端口 self.server_activate() #elf.socket.listen(self.request_queue_size) 监听 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 selector. """ 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 OSError: 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 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 selector 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) - service_actions() - 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: # 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. with _ServerSelector() as selector: selector.register(self, selectors.EVENT_READ) while not self.__shutdown_request: ready = selector.select(poll_interval) if ready: self._handle_request_noblock() self.service_actions() 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() def service_actions(self): """Called by the serve_forever() loop. May be overridden by a subclass / Mixin to implement any code that needs to be run during the loop. """ pass # The distinction between handling, getting, processing and finishing a # request is fairly arbitrary. Remember: # # - handle_request() is the top-level call. It calls selector.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) if timeout is not None: deadline = time() + timeout # Wait until a request arrives or the timeout expires - the loop is # necessary to accommodate early wakeups due to EINTR. with _ServerSelector() as selector: selector.register(self, selectors.EVENT_READ) while True: ready = selector.select(timeout) if ready: return self._handle_request_noblock() else: if timeout is not None: timeout = deadline - time() if timeout < 0: return self.handle_timeout() def _handle_request_noblock(self): """Handle one request, without blocking. I assume that selector.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() #self.socket.accept()所以request相当于连接conn except OSError: 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) else: 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‘, end=‘ ‘) print(client_address) import traceback traceback.print_exc() # XXX But this goes to stderr! print(‘-‘*40)
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 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 other arbitrary instance variables. """ 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
参考:
https://docs.python.org/3.6/library/socket.html
https://www.cnblogs.com/linhaifeng/articles/6129246.html
标签:参数 数据包 start log mat eth should 匹配 面向连接
原文地址:https://www.cnblogs.com/shenjianping/p/11610001.html