socketserver

Python socketserver 模块教程

socketserver 模块提供了一组简单易用的类，用于编写网络服务器。它封装了底层的 socket API，使得开发 TCP、UDP 服务器变得更加容易。通过使用 socketserver 模块，开发者可以专注于处理客户端请求，而不必关心底层的网络通信细节。

官方文档链接

Python socketserver官方文档

基本功能

1. TCP 服务器

以下是一个简单的 TCP 服务器示例，该服务器接受客户端连接并回显客户端发送的数据：

import socketserver

class MyTCPHandler(socketserver.BaseRequestHandler):
    def handle(self):
        self.data = self.request.recv(1024).strip()
        print(f"{self.client_address[0]} wrote:")
        print(self.data)
        self.request.sendall(self.data)

if __name__ == "__main__":
    HOST, PORT = "localhost", 9999
    with socketserver.TCPServer((HOST, PORT), MyTCPHandler) as server:
        print("Server started at {}:{}".format(HOST, PORT))
        server.serve_forever()

2. UDP 服务器

以下是一个简单的 UDP 服务器示例，该服务器接受客户端数据并回显：

import socketserver

class MyUDPHandler(socketserver.BaseRequestHandler):
    def handle(self):
        data = self.request[0].strip()
        socket = self.request[1]
        print(f"{self.client_address[0]} wrote:")
        print(data)
        socket.sendto(data, self.client_address)

if __name__ == "__main__":
    HOST, PORT = "localhost", 9999
    with socketserver.UDPServer((HOST, PORT), MyUDPHandler) as server:
        print("Server started at {}:{}".format(HOST, PORT))
        server.serve_forever()

高级功能

1. 多线程服务器

为了处理多个客户端连接，可以使用 ThreadingMixIn 来创建多线程服务器：

import socketserver
import threading

class MyTCPHandler(socketserver.BaseRequestHandler):
    def handle(self):
        self.data = self.request.recv(1024).strip()
        print(f"{self.client_address[0]} wrote:")
        print(self.data)
        self.request.sendall(self.data)

class ThreadedTCPServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
    pass

if __name__ == "__main__":
    HOST, PORT = "localhost", 9999
    with ThreadedTCPServer((HOST, PORT), MyTCPHandler) as server:
        ip, port = server.server_address
        print("Server started at {}:{}".format(ip, port))
        server.serve_forever()

2. 多进程服务器

为了处理多个客户端连接，可以使用 ForkingMixIn 来创建多进程服务器：

import socketserver
import os

class MyTCPHandler(socketserver.BaseRequestHandler):
    def handle(self):
        self.data = self.request.recv(1024).strip()
        print(f"{self.client_address[0]} wrote:")
        print(self.data)
        self.request.sendall(self.data)

class ForkingTCPServer(socketserver.ForkingMixIn, socketserver.TCPServer):
    pass

if __name__ == "__main__":
    HOST, PORT = "localhost", 9999
    with ForkingTCPServer((HOST, PORT), MyTCPHandler) as server:
        print("Server started at {}:{}".format(HOST, PORT))
        server.serve_forever()

3. 覆盖默认行为

可以覆盖 socketserver 的一些默认行为，例如，设置请求超时或处理服务器关闭逻辑：

import socketserver

class MyTCPHandler(socketserver.BaseRequestHandler):
    def handle(self):
        self.data = self.request.recv(1024).strip()
        print(f"{self.client_address[0]} wrote:")
        print(self.data)
        self.request.sendall(self.data)

class MyTCPServer(socketserver.TCPServer):
    def server_activate(self):
        self.timeout = 5
        super().server_activate()
        print("Server activated with timeout:", self.timeout)

    def server_close(self):
        super().server_close()
        print("Server closed")

if __name__ == "__main__":
    HOST, PORT = "localhost", 9999
    with MyTCPServer((HOST, PORT), MyTCPHandler) as server:
        print("Server started at {}:{}".format(HOST, PORT))
        server.serve_forever()

使用示例

以下是一个完整的示例，展示了如何使用 socketserver 模块创建一个简单的回显服务器：

1. TCP 服务器

import socketserver

class MyTCPHandler(socketserver.BaseRequestHandler):
    def handle(self):
        self.data = self.request.recv(1024).strip()
        print(f"{self.client_address[0]} wrote:")
        print(self.data)
        self.request.sendall(self.data)

if __name__ == "__main__":
    HOST, PORT = "localhost", 9999
    with socketserver.TCPServer((HOST, PORT), MyTCPHandler) as server:
        print("Server started at {}:{}".format(HOST, PORT))
        server.serve_forever()

2. UDP 服务器

import socketserver

class MyUDPHandler(socketserver.BaseRequestHandler):
    def handle(self):
        data = self.request[0].strip()
        socket = self.request[1]
        print(f"{self.client_address[0]} wrote:")
        print(data)
        socket.sendto(data, self.client_address)

if __name__ == "__main__":
    HOST, PORT = "localhost", 9999
    with socketserver.UDPServer((HOST, PORT), MyUDPHandler) as server:
        print("Server started at {}:{}".format(HOST, PORT))
        server.serve_forever()

3. 客户端

import socket

def tcp_client():
    HOST, PORT = "localhost", 9999
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.connect((HOST, PORT))
        sock.sendall(b"Hello, World!")
        response = sock.recv(1024)
        print(f"Received: {response}")

def udp_client():
    HOST, PORT = "localhost", 9999
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
        sock.sendto(b"Hello, World!", (HOST, PORT))
        response, _ = sock.recvfrom(1024)
        print(f"Received: {response}")

if __name__ == "__main__":
    tcp_client()
    udp_client()

总结

socketserver 模块提供了一组方便的类，用于快速编写网络服务器。通过使用这些类，开发者可以专注于处理客户端请求，而不必关心底层的网络通信细节。通过本文介绍的基本功能和高级功能，开发者可以轻松上手并熟练运用 socketserver 进行各种网络编程任务。更多详细信息和示例请参考官方文档。