SocketServer源码学习

Posted 2020-10-31 coder

SocketServer其实是对socket更高级的封装正如官网上说的：
The socketserver module simplifies the task of writing network servers.

我们可以先打开以下SocketServer的源码，看一下源码中整体的框架

从上图我们可以看出SocketServer主要被抽象为两个主要的类：
BaseServer类,用于处理连接相关的网络操作
BaseRequestHandler类，用于实际处理数据相关的操作

SocketServer还提供了两个MixIn类：ThreadingMinxIn和ForkingMixinl
用于扩展server，实现多进程和多线程

下面从会从这几个主要的类开始做一个整体的分析，了解SocketServer的处理流程

 

BaseServer

先列一下这里所包含的方法：
server_activate
serve_forever
shutdown
service_actions
handle_request
handlerequest_noblock
handle_timeout
verify_request
process_request
server_close
finish_request
shutdown_request
close_request
handle_error

先看一下BaseServer的初始化函数，其实并没有过多的参数，主要就是实现了创建server对象，并初始化server地址和处理请求的类：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

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

当创建server对象之后，我们会使用server对象开启一个无限循环，即调用serve_forever,
下面是它的源码，接受一个参数poll_interval，用于表示select轮询的时间，然后进入一个死循环，用select方法进行网络IO的监听，这里通过调用selector.register(self, selectors.EVENT_READ)进行了注册，当ready有返回是，表示有IO连接或者数据，这个时候会调用_handle_request_noblock
接着看一下_handle_request_noblock源码

handlerequest_noblock

源码内容如下：

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:
            self.handle_error(request, client_address)
            self.shutdown_request(request)
    else:
        self.shutdown_request(request)

handlerequest_noblock方法即开始处理一个请求，并且是非阻塞。该方法通过get_request方法获取连接，具体的实现在其子类。一旦得到了连接，调用verify_request方法验证请求。验证通过，即调用process_request处理请求。如果中途出现错误，则调用handle_error处理错误，以及shutdown_request结束连接。

而verify_request中默认直接返回True，所以当验证通过后讲调用process_request

process_request

源码内容如下：

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)

就像源码中描述的那样：Overridden by ForkingMixIn and ThreadingMixIn.
process_request方法是mixin的入口，MixIn子类通过重写该方法，进行多线程或多进程的配置。调用finish_request完成请求的处理，同时调用shutdown_request结束请求。继续查看finish_request

finish_request

源码内容如下：

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

关于请求的部分到这里就已经处理完毕，接下来是要对数据的处理，finish_request方法将会处理完毕请求。创建requestHandler对象，并通过requestHandler做具体的处理。

BaseRequestHandler

就像我们前面说的：
BaseServer类,用于处理连接相关的网络操作
BaseRequestHandler类，用于实际处理数据相关的操作

还是从初始化函数里看源码：

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

该类会处理每一个请求。初始化对象的时候，设置请求request对象。然后调用setup方法，子类会重写该方法，用于处理socket连接。接下来的将是handler和finish方法。所有对请求的处理，都可以重写handler方法。

SocketServer的一个服务端的简单例子

直接上代码了：

import socketserver


class MyTCPHandler(socketserver.BaseRequestHandler):
    def handle(self):
        while True:
            try:
                self.data = self.request.recv(1024).strip()
                print("{} wrote:".format(self.client_address))
                print(self.data)
                self.request.sendall(self.data.upper())
            except ConnectionResetError as e:
                print(e)
                break

if __name__ == \'__main__\':
    host,port = "127.0.0.1",9999
    server = socketserver.TCPServer((host,port),MyTCPHandler)
    server.serve_forever()

通过对源码的分析我们已经知道了对数据的处理部分都是在BaseRequestHandler这个类中，而我们的主要处理逻辑也就是在这部分，所以我继承了这个类，重写了handle方法