socketserver
socketserver内部使用IO多路复用以及“多线程”和“多进程”,从而实现并发处理多个客户端请求的scoket服务端。即,每个客户端请求连接到服务器时,socket服务端都会在服务器是创建一个“线程”或“进程”专门负责处理当前客户端的所有请求。
ThradingTCPServer
ThradingTCPServer实现的socket服务器内部会为每个client创建一个“线程”,该线程用来客户端进行交互
1、ThreadingTCPServer基础
使用ThreadingTCPServer:
- 创建一个继承自 SocketServer.BaseRequestHandler 的类
- 类中必须定义一个名称为 handle 的方法
- 启动ThreadingTCPServer
使用
server端
import socketserver class MyServer(socketserver.BaseRequestHandler): # $ 必须继承BaseRequestHandler print('Myserver start'.center(50, "-")) def handle(self): # $ 必须有handle方法 print('New connection:', self.client_address) while True: data = self.request.recv(1024) if not data: break print( 'recv Client {} data {}'.format( self.client_address, data.decode())) self.request.send(data) if __name__ == '__main__': server = socketserver.ThreadingTCPServer( ('127.0.0.1', 8009), MyServer) # $ 实现多线程的socket server.serve_forever() # $ 当前连接断开不会出现关闭或报错,可以与其他客户端继续连接
client端:
import socket ip_port = ('127.0.0.1',8009) sk = socket.socket() sk.connect(ip_port) while True: raw = input('client input >> ').strip() sk.send(raw.encode("utf-8")) msg = sk.recv(1024) print(msg.decode("utf-8")) sk.close()
输出
------------------Myserver start------------------ New connection: ('127.0.0.1', 13560) recv Client ('127.0.0.1', 13560) data hello New connection: ('127.0.0.1', 13561) recv Client ('127.0.0.1', 13561) data world ---------------client1: client input >> hello hello client input >> ---------------client2: client input >> world world client input >>
socketserver分析
socketserver中包含了两种类,一种为服务类(server class),一种为请求处理类(request handle class)。前者提供了许多方法:像绑定,监听,运行…… (也就是建立连接的过程) 后者则专注于如何处理用户所发送的数据(也就是事务逻辑)。
服务类继承层次关系
BaseServer不直接对外服务。
TCPServer针对TCP套接字流
UDPServer针对UDP数据报套接字
UnixStreamServer和UnixDatagramServer针对UNIX域套接字
ThreadingTCPServer 源码分析
BaseServer
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
TCPServer
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()
ThreadingMixIn
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) except Exception: self.handle_error(request, client_address) finally: 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()
RequestHandler
class ThreadingTCPServer(ThreadingMixIn, TCPServer): pass 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
请求处理类有三种方法:
setup
()-
Called before the
handle()
method to perform any initialization actions required. The default implementation does nothing.也就是在handle()之前被调用,主要的作用就是执行处理请求之前的初始化相关的各种工作。默认不会做任何事。(如果想要让其做一些事的话,就要程序员在自己的请求处理器中覆盖这个方法(因为一般自定义的请求处理器都要继承python中提供的BaseRequestHandler,ps:下文会提到的),然后往里面添加东西即可)
handle
()-
This function must do all the work required to service a request. The default implementation does nothing. Several instance attributes are available to it; the request is available as
self.request
; the client address asself.client_address
; and the server instance asself.server
, in case it needs access to per-server information.The type of
self.request
is different for datagram or stream services. For stream services,self.request
is a socket object; for datagram services,self.request
is a pair of string and socket.handle()的工作就是做那些所有与处理请求相关的工作。默认也不会做任何事。他有数个实例参数:self.request self.client_address self.server
finish
()-
Called after the
handle()
method to perform any clean-up actions required. The default implementation does nothing. Ifsetup()
raises an exception, this function will not be called.在handle()方法之后会被调用,他的作用就是执行当处理完请求后的清理工作,默认不会做任何事
内部调用流程为:
- 启动服务端程序
- 执行 TCPServer.__init__ 方法,创建服务端Socket对象并绑定 IP 和 端口
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()
执行 BaseServer.__init__ 方法,将自定义的继承自SocketServer.BaseRequestHandler 的类 MyRequestHandle赋值给 self.RequestHandlerClass
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
执行 BaseServer.server_forever 方法,While 循环一直监听是否有客户端请求到达 ...
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() except OSError: return if self.verify_request(request, client_address): try: self.process_request(request, client_address) except Exception: self.handle_error(request, client_address) self.shutdown_request(request) except: self.shutdown_request(request) raise else: self.shutdown_request(request)
执行 ThreadingMixIn.process_request 方法,创建一个 “线程” 用来处理请求
执行 ThreadingMixIn.process_request_thread 方法
# 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) except Exception: self.handle_error(request, client_address) finally: 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()
执行 BaseServer.finish_request 方法,执行 self.RequestHandlerClass() 即:执行 自定义 MyRequestHandler 的构造方法(自动调用基类BaseRequestHandler的构造方法,在该构造方法中又会调用 MyRequestHandler的handle方法)
finish_request方法中执行了self.RequestHandlerClass(request, client_address, self)。self.RequestHandlerClass是什么呢?
self.RequestHandlerClass = RequestHandlerClass(就在__init__方法中)。所以finish_request方法本质上就是创建了一个服务处理实例RequestHandlerClass
RequestHandlerClass继承了BaseRequestHandler类,并实现handle方法,在实例化RequestHandlerClass时,会执行handle方法,处理用户所发送的数据(也就是事务逻辑)。
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()
时序图
附录
class SocketServer.BaseServer:这是模块中的所有服务器对象的超类。它定义了接口,如下所述,但是大多数的方法不实现,在子类中进行细化。 BaseServer.fileno():返回服务器监听套接字的整数文件描述符。通常用来传递给select.select(), 以允许一个进程监视多个服务器。 BaseServer.handle_request():处理单个请求。处理顺序:get_request(), verify_request(), process_request()。如果用户提供handle()方法抛出异常,将调用服务器的handle_error()方法。如果self.timeout内没有请求收到, 将调用handle_timeout()并返回handle_request()。 BaseServer.serve_forever(poll_interval=0.5): 处理请求,直到一个明确的shutdown()请求。每poll_interval秒轮询一次shutdown。忽略self.timeout。如果你需要做周期性的任务,建议放置在其他线程。 BaseServer.shutdown():告诉serve_forever()循环停止并等待其停止。python2.6版本。 BaseServer.address_family: 地址家族,比如socket.AF_INET和socket.AF_UNIX。 BaseServer.RequestHandlerClass:用户提供的请求处理类,这个类为每个请求创建实例。 BaseServer.server_address:服务器侦听的地址。格式根据协议家族地址的各不相同,请参阅socket模块的文档。 BaseServer.socketSocket:服务器上侦听传入的请求socket对象的服务器。 服务器类支持下面的类变量: BaseServer.allow_reuse_address:服务器是否允许地址的重用。默认为false ,并且可在子类中更改。 BaseServer.request_queue_size 请求队列的大小。如果单个请求需要很长的时间来处理,服务器忙时请求被放置到队列中,最多可以放request_queue_size个。一旦队列已满,来自客户端的请求将得到 “Connection denied”错误。默认值通常为5 ,但可以被子类覆盖。 BaseServer.socket_type:服务器使用的套接字类型; socket.SOCK_STREAM和socket.SOCK_DGRAM等。 BaseServer.timeout:超时时间,以秒为单位,或 None表示没有超时。如果handle_request()在timeout内没有收到请求,将调用handle_timeout()。 下面方法可以被子类重载,它们对服务器对象的外部用户没有影响。 BaseServer.finish_request():实际处理RequestHandlerClass发起的请求并调用其handle()方法。 常用。 BaseServer.get_request():接受socket请求,并返回二元组包含要用于与客户端通信的新socket对象,以及客户端的地址。 BaseServer.handle_error(request, client_address):如果RequestHandlerClass的handle()方法抛出异常时调用。默认操作是打印traceback到标准输出,并继续处理其他请求。 BaseServer.handle_timeout():超时处理。默认对于forking服务器是收集退出的子进程状态,threading服务器则什么都不做。 BaseServer.process_request(request, client_address) :调用finish_request()创建RequestHandlerClass的实例。如果需要,此功能可以创建新的进程或线程来处理请求,ForkingMixIn和ThreadingMixIn类做到这点。常用。 BaseServer.server_activate():通过服务器的构造函数来激活服务器。默认的行为只是监听服务器套接字。可重载。 BaseServer.server_bind():通过服务器的构造函数中调用绑定socket到所需的地址。可重载。 BaseServer.verify_request(request, client_address):返回一个布尔值,如果该值为True ,则该请求将被处理,反之请求将被拒绝。此功能可以重写来实现对服务器的访问控制。默认的实现始终返回True。client_address可以限定客户端,比如只处理指定ip区间的请求。 常用。
这个几个服务类都是同步处理请求的:一个请求没处理完不能处理下一个请求。要想支持异步模型,可以利用多继承让server类继承ForkingMixIn 或 ThreadingMixIn mix-in classes。
ForkingMixIn利用多进程(分叉)实现异步。
ThreadingMixIn利用多线程实现异步。
参考:https://www.cnblogs.com/MnCu8261/p/5546823.html