python3 socketserver模块网络服务编程框架

悠悠 2021-07-28 11:19 476阅读 0赞

socket编程过于底层，编程虽然有套路，但是想要写出健壮的代码还是比较困难的，所以很多语言都对socket底层 API进行封装，Python的封装就是——socketserver模块。它是网络服务编程框架，便于企业级快速开发。

1.  **类的继承关系**
    
        +------------+
        | BaseServer |
        +------------+
            |
            v
        +-----------+        +------------------+
        | TCPServer |------->| UnixStreamServer |
        +-----------+        +------------------+
            |
            v
        +-----------+        +--------------------+
        | UDPServer |------->| UnixDatagramServer |
        +-----------+        +--------------------+
2.  SocketServer简化了网络服务器的编写
    
     *  它有4个同步类：
        
        1.  TCPServer
        2.  UDPServer
        3.  UnixStreamServer
        4.  UnixDatagramServer
     *  两个Mixin类，用来支持异步。
        
         *  ForkingMixIn
         *  ThreadingMixIn
     *  组合得到
        
         *  class ForkingUDPServer(ForkingMixIn,UDPServer):pass
         *  class ForkingTCPServer(ForkingMixIn,TCPServer):pass
         *  class ThreadingUDPServer(ThreadingMixIn,UDPServer):pass
         *  class ThreadingTCPServer(ThreadingMixIn,TCPServer):pass

*  fork是创建多进程，thread是创建多线程。
 *  fork需要操作系统支持，Windows不支持。
 *  ThreadingUDPServer与ThreadingTCPServer类中的特有属性：
    
    1.  daemon\_threads=False \#默认值是False表示创建的线程都不是daemon线程，改为True表示创建的所有线程都是daemon线程
    2.  block\_on\_close=False \#默认值为Fasle，如果为True，可以设置为守护线程3.7版本可以用

## 编程接口 ##

1.  socketserver.BaseServer(server\_address,RequestHandlerClass) \#实例化一个服务器
    
     *  server\_address \#服务器绑定的地址信息，是一个元组(ip,prost)
     *  RequestHandlerClass \#必须是BaseRequestHandler的一个子类。
     *  在BaseServer中原码代码如下：

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
    
    # 处理请求的方法，会实例化一个RequestHandlerClass对象
    def finish_request(self, request, client_address):
        """Finish one request by instantiating RequestHandlerClass."""
        self.RequestHandlerClass(request, client_address, self)

1.  BaseServer中定义的接口常用属性和方法

<table> 
 <thead> 
  <tr> 
   <th>方法</th> 
   <th>含义</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><strong>server_address</strong></td> 
   <td>服务器正在监听的地址和端口，在不同协议格式不一样。Internet协议上是一个元组(“127.0.0.1”,80)</td> 
  </tr> 
  <tr> 
   <td><strong>socket</strong></td> 
   <td>服务器正在监听的套接字对象。socket</td> 
  </tr> 
  <tr> 
   <td>request_queue_size</td> 
   <td>请求队列的大小。如果处理单个请求需要很长时间，那么在服务器繁忙时到达的任何请求都会被放入队列中，直到request_queue_size请求为止。一旦队列满了，来自客户机的进一步请求将得到一个“连接被拒绝”错误。默认值通常是5，但是可以被子类覆盖。</td> 
  </tr> 
  <tr> 
   <td>address_family</td> 
   <td>服务器套接字所属的协议族。常见的例子是套接字。AF_INET socket.AF_UNIX。</td> 
  </tr> 
  <tr> 
   <td>socket_type</td> 
   <td>服务器使用的套接字类型;套接字。SOCK_STREAM套接字。SOCK_DGRAM是两个常见的值。</td> 
  </tr> 
  <tr> 
   <td>timeout</td> 
   <td>超时持续时间，以秒为单位度量，如果不需要超时，则为None。如果handle_request()在超时期间没有收到传入的请求，则调用handle_timeout()方法。</td> 
  </tr> 
  <tr> 
   <td><strong>handle_request()</strong></td> 
   <td><strong>处理单个请求，同步执行</strong><br> 这个函数按顺序调用以下方法:get_request()、verify_request()和process_request()。如果处理程序类的用户提供的handle()方法引发异常，将调用服务器的handle_error()方法。如果在超时秒内没有收到任何请求，那么将调用handle_timeout()并返回handle_request()。</td> 
  </tr> 
  <tr> 
   <td><strong>server_forever(poll_interval=0.5)</strong></td> 
   <td><strong>异步执行，处理请求</strong>。每隔poll_interval秒轮询一次。<br> 忽略timeout属性，还会调用service_actions()，在ForkingMixIn的子类中定义，可以用来清理僵尸进程。</td> 
  </tr> 
  <tr> 
   <td><strong>shutdown()</strong></td> 
   <td>告诉serve_forever循环停止。并等待他结束。</td> 
  </tr> 
  <tr> 
   <td><strong>server_close()</strong></td> 
   <td>关闭服务器</td> 
  </tr> 
  <tr> 
   <td>finish_request(request,client_address)</td> 
   <td>通过实例化RequestHandlerClass并调用它的handle()方法来处理请求</td> 
  </tr> 
  <tr> 
   <td>server_bind()</td> 
   <td>由服务器的构造函数调用，以将套接字绑定到所需的地址。可能会被覆盖。</td> 
  </tr> 
  <tr> 
   <td>verify_request(request,client_address)</td> 
   <td>必须返回一个布尔值;如果值为True，请求将被处理，如果值为False，请求将被拒绝。可以重写此函数来实现服务器的访问控制。默认实现总是返回True。</td> 
  </tr> 
 </tbody> 
</table>

### BaseRequestHandler类 ###

1.  是和用户连接的用户请求处理类的基类，
2.  `BaseRequestHandler(request,client_address,server)` \#构造函数
    
     *  request \#是和客户端的连接的socket对象
     *  client\_address \#是客户端地址
     *  server \#是TCPServer实例本身
3.  服务端Server实例接收用户请求后，最后会实例化这个类。它被初始化时，送入3个构造参数：request, client\_address, server自身 以后就可以在BaseRequestHandler类的实例上使用以下属性：
    
     *  self.request是和客户端的连接的socket对象
     *  self.server是TCPServer实例本身
     *  self.client\_address是客户端地址
4.  这个类在初始化的时候，它会依次调用3个方法。子类可以覆盖这些方法。

class 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()
    
        def setup(self): #每一个连接初始化
            pass
    
        def handle(self): #每一次请求处理
            pass
    
        def finish(self): #每一个连接清理
            pass

测试代码

import socketserver
    import socket
    
    class MyBaseRequestHandle(socketserver.BaseRequestHandler):
    
        def setup(self):
            super().setup() #可以不调用父类的setup()方法，父类的setup方法什么都没做
            print("----setup方法被执行-----")
    
        def handle(self):
            super().handle() #可以不调用父类的handler(),方法，父类的handler方法什么都没做
            print("-------handler方法被执行----")
            print(self.server)
            print(self.request) #服务
            print(self.client_address) #客户端地址
            print(self.__dict__)
            print("- "*30)
            print(self.server.__dict__)
            print("- "*30)
            sk:socket.socket = self.request
            data = sk.recv(1024)
            print(data)
            sk.send("{}-{}".format(sk.getpeername(),data).encode())
    
            print("----------handler end ----------")
    
        def finish(self):
            super().finish() #可以不调用父类的finish(),方法，父类的finish方法什么都没做
            print("--------finish方法被执行---")
    
    laddr = "127.0.0.1",3999
    tcpserver = socketserver.TCPServer(laddr,MyBaseRequestHandle) #注意：参数是MyBaseRequestHandle
    tcpserver.handle_request() #只接受一个客户端连接
    # tcpserver.serve_forever() #永久循环执行,可以接受多个客户端连接

![socketserver\_001.jpg][socketserver_001.jpg]

每个不同的连接上的请求过来后，生成这个连接的socket对象即self.request，客户端地址是self.client\_address。

将ThreadingTCPServer换成TCPServer，当每个客户端连接进来就会创建一个新的线程。

ThreadingTCPServer是异步的，可以同时处理多个连接。

TCPServer是同步的，一个连接处理完了，即一个连接的handle方法执行完了，才能处理另一个连接，且只有主线程。

## 总结 ##

创建服务器需要几个步骤：

1.  从BaseRequestHandler类派生出子类，并覆盖其handle()方法来创建请求处理程序类，此方法将处理传入请求
2.  实例化一个服务器类，传参服务器的地址和请求处理类
3.  调用服务器实例的handle\_request()或serve\_forever()方法
4.  调用server\_close()关闭套接字

## 实现EchoServer ##

**Echo**:来声明消息，回显什么消息，即客户端发来什么消息，返回什么消息

import logging
    import sys
    import socketserver
    import socket
    import threading
    
    logging.basicConfig(format="%(asctime)s %(thread)d %(threadName)s %(message)s",stream=sys.stdout,level=logging.INFO)
    
    class Handler(socketserver.BaseRequestHandler):
        def setup(self):
            super().setup()
            self.event = threading.Event()
            logging.info("新加入了一个连接{}".format(self.client_address))
    
        def handle(self):
            super().handle()
            sk:socket.socket = self.request
            while not self.event.is_set():
                try:
                    data = sk.recv(1024).decode()
                except Exception as e:
                    logging.info(e)
                    break
    
                logging.info(data)
                msg = "{}-{}".format(self.client_address,data).encode()
                sk.send(msg)
    
        def finish(self):
            super().finish()
            self.event.set()
            self.request.close()
    
    if __name__ == "__main__":
        server = socketserver.ThreadingTCPServer(("127.0.0.1",3999),Handler)
        threading.Thread(target=server.serve_forever,name="server").start()
        while True:
            cmd = input(">>>")
            if cmd.strip() == "quit":
                server.server_close()
                break
            logging.info(threading.enumerate())

**实战，使用socketserver实现群聊的server**

使用ThreadingTCPServer改写ChatServer

使用BaseRequestHandler定义Handler

import logging
    import sys
    import socketserver
    import socket
    import threading
    
    logging.basicConfig(format="%(asctime)s %(thread)d %(threadName)s %(message)s",stream=sys.stdout,level=logging.INFO)
    log = logging.getLogger()
    
    class Handler(socketserver.BaseRequestHandler):
        lock = threading.Lock()
        clients = {}
    
        def setup(self):
            super().setup()
            self.event = threading.Event()
            with self.lock:
                self.clients[self.client_address] = self.request
            log.info("新加入了一个连接{}".format(self.client_address))
    
        def handle(self):
            super().handle()
            sock:socket.socket = self.request
            while not self.event.is_set():
                try:
                    data = sock.recv(1024)
                except Exception as e:
                    log.error(e)
                    data = b""
                log.info(data)
                if data == b"by" or data == b"":
                    break
                msg = "service:{}-->{}".format(self.client_address, data).encode()
                expc = []  # 记录sock出错时对应的clients
                with self.lock:
                    for c, sk in self.clients.items():
                        try:
                            sk.send(msg)  # 可能在发送消息是就出错
                        except:
                            expc.append(c)
                    for c in expc:
                        self.clients.pop(c)
    
        def finish(self):
            super().finish()
            self.event.set()
            with self.lock:
                if self.client_address in self.clients:
                    self.clients.pop(self.client_address)
            self.request.close()
            log.info("{}退出了".format(self.client_address))
    
    if __name__ == "__main__":
        server = socketserver.ThreadingTCPServer(("127.0.0.1",3999),Handler)
        server.daemon_threads = True  #设置所有创建的线程都为Daemo线程
        threading.Thread(target=server.serve_forever,name="server",daemon=True).start()
        while True:
            cmd = input(">>>")
            if cmd.strip() == "quit":
                server.shutdown() #告诉serve_forever循环停止。
                server.server_close()
                break
            logging.info(threading.enumerate())

使用StreamRequestHandler定义handler

import logging
    import sys
    import socketserver
    import socket
    import threading
    
    logging.basicConfig(format="%(asctime)s %(thread)d %(threadName)s %(message)s",stream=sys.stdout,level=logging.INFO)
    log = logging.getLogger()
    
    class Handler(socketserver.StreamRequestHandler):
        lock = threading.Lock()
        clients = {}
    
        def setup(self):
            super().setup()
            self.event = threading.Event()
            with self.lock:
                self.clients[self.client_address] = self.request
            log.info("新加入了一个连接{}".format(self.client_address))
    
        def handle(self):
            super().handle()
            import io
            rfile:io.TextIOWrapper= self.rfile
            while not self.event.is_set():
                try:
                    data = rfile.read1(1024) #类似于sock.recv(1024)
                    # data = rfile.readline() #行读取
                except Exception as e:
                    log.error(e)
                    data = b""
                log.info(data)
                if data == b"by" or data == b"":
                    break
                msg = "service:{}-->{}".format(self.client_address, data).encode()
                expc = []  # 记录sock出错时对应的clients
                with self.lock:
                    for c, sk in self.clients.items():
                        try:
                            sk.send(msg)  # 可能在发送消息是就出错
                        except:
                            expc.append(c)
                    for c in expc:
                        self.clients.pop(c)
    
        def finish(self):
            super().finish()
            self.event.set()
            with self.lock:
                if self.client_address in self.clients:
                    self.clients.pop(self.client_address)
            self.request.close()
            log.info("{}退出了".format(self.client_address))
    
    if __name__ == "__main__":
        server = socketserver.ThreadingTCPServer(("127.0.0.1",3999),Handler)
        server.daemon_threads = True  #设置所有创建的线程都为Daemo线程
        threading.Thread(target=server.serve_forever,name="server",daemon=True).start()
        while True:
            cmd = input(">>>")
            if cmd.strip() == "quit":
                server.shutdown() #告诉serve_forever循环停止。
                server.server_close()
                break
            logging.info(threading.enumerate())

**总结**

1.  为每一个连接提供RequestHandlerClass类实例，依次调用setup、handle、ﬁnish方法，且使用了try…ﬁnally结构 保证ﬁnish方法一定能被调用。这些方法依次执行完成，如果想维持这个连接和客户端通信，就需要在handle函数 中使用循环。
2.  socketserver模块提供的不同的类，但是编程接口是一样的，即使是多进程、多线程的类也是一样，大大减少了编 程的难度。
3.  将socket编程简化，只需要程序员关注数据处理本身，实现Handler类就行了。这种风格在Python十分常见。

[socketserver_001.jpg]: /images/20210728/db7718f72dd74e7d9cdec71d29e56785.png