Python之路(第三十三篇) 網路程式設計：socketserver深度解析
阿新 • • 發佈：2018-11-23
模組介紹
socketserver是標準庫中的一個高階模組，用於網路客戶端與伺服器的實現。(version = "0.4")
在python2中寫作SocketServer,在python3中寫作socketserver。
socoketserver兩個主要的類，一個是Server類，用於處理連線相關的網路操作，另外一個則是RequestHandler類，用於處理資料相關的操作，解決通訊問題。並且提供兩個MixIn 類，用於擴充套件 Server，實現多程序或多執行緒。
最基類的是伺服器類BaseServer類，其中定義了相關的方法，不能直接使用這個類，只能用來繼承，在子類中有倆，是作為同步伺服器類使用，TCPServer和UDPServer，這兩個類主要是和socket程式設計的時候是相同的，也就是會阻塞連線。TCPServer有一個子類為UNIXStreamServer，在UDPServer有一個子類為UnixDatagramServer，這兩個是用在UNIX系統中的。
兩個混合類，一個是ForkingMixin，主要是用fork的（在linux系統中使用產生程序，在Windows系統中無法使用fork產生程序），產生一個新的程序去處理；一個是ThreadingMixin，產生一個新的執行緒，主要是用來提供非同步處理（簡單理解為同時處理多個事情）的能力，其餘tcpserver和udpserver組合，又產生了新的四個類，從而提供非同步處理的能力。
socketserver框架
在python中，socketserver是一個已整合的模組，它有以下特點：
-
可用SocketServer框架建立TCP和UDP伺服器。
-
在後臺為你做好每一個基礎步驟。
socketserver模組的用法
-
必須是BaseRequestHandler的子類
-
重寫handle()函式
-
呼叫serve_forever處理客戶端程式。
-
對TCP 服務端來說
-
self.request是客戶端Socket物件
-
self.client_address是客戶端的詳細地址
-
socketserver模組中的各個類
伺服器類
5種類型：BaseServer，TCPServer，UnixStreamServer，UDPServer，UnixDatagramServer。 注意：BaseServer不直接對外服務。
請求類
3種類型：BaseRequestHandler，StreamRequestHandler，DatagramRequestHandler
minln類
ForkingMinIn ，ThreadingMinIn
混合類
ThreadingUDPServer, ThreadingTCPServer，ForkingUDPServer,ForkingTCPServer,
ThreadingUnixStreamServer，ThreadingUnixDatagramServer
-
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區間的請求。 常用。
請求處理器
處理器接收資料並決定如何操作。它負責在socket層之上實現協議（i.e., HTTP, XML-RPC, or AMQP)，讀取資料，處理並寫反應。可以過載的方法如下：
-
setup(): 準備請求處理. 預設什麼都不做，StreamRequestHandler中會建立檔案類似的物件以讀寫socket.
-
handle(): 處理請求。解析傳入的請求，處理資料，併發送響應。預設什麼都不做。常用變數：self.request，self.client_address，self.server。
-
finish(): 環境清理。預設什麼都不做，如果setup產生異常，不會執行finish。
通常只需要過載handle。self.request的型別和資料報或流的服務不同。對於流服務，self.request是socket 物件；對於資料報服務，self.request是字串和socket 。可以在子類StreamRequestHandler或DatagramRequestHandler中過載，重寫setup()和finish() ，並提供self.rfile和self.wfile屬性。 self.rfile和self.wfile可以讀取或寫入，以獲得請求資料或將資料返回到客戶端。
socketserver模組原始碼翻譯
"""Generic socket server classes.
# 通用socket server 類
This module tries to capture the various aspects of defining a server:
# 該模組盡力從各種不同的方面定義server：
For socket-based servers:
- address family:
- AF_INET{,6}: IP (Internet Protocol) sockets (default)
- AF_UNIX: Unix domain sockets
- others, e.g. AF_DECNET are conceivable (see <socket.h>
- socket type:
- SOCK_STREAM (reliable stream, e.g. TCP)
- SOCK_DGRAM (datagrams, e.g. UDP)
# 對於socket-based servers：(對於socketserver類中的服務類的基類有下面內容)
# -- address family：(地址家族)
# - AF_INET{,6}: IP socket (default)
# - AF_UNIX: Unix domain sockets
# - others, 如 AF_DECNET (見<socket.h>) (不常用)
# -- socket type：(套接字型別)
# - SOCK_STREAM (可靠連線 TCP)
# - SOCK_DGRAM (UDP)
For request-based servers (including socket-based):
- client address verification before further looking at the request
(This is actually a hook for any processing that needs to look
at the request before anything else, e.g. logging)
- how to handle multiple requests:
- synchronous (one request is handled at a time)
- forking (each request is handled by a new process)
- threading (each request is handled by a new thread)
# 對於request-based servers：（對於通訊請求服務類的基類）
# -- client address在發出進一步的請求之前需要認證（這實際上把所有需要發出請求的程序在通過認證之前給阻塞住了）
# -- 如何處理多請求：(多程序多執行緒)
# - 同步 （一次只能處理一個請求）
# - forking （fork一個新的程序來處理一個請求）
# - threading （建立一個新的執行緒來處理一個請求）
The classes in this module favor the server type that is simplest to
write: a synchronous TCP/IP server. This is bad class design, but
save some typing. (There's also the issue that a deep class hierarchy
slows down method lookups.)
# 在這個模組的各種類中，最簡單的伺服器型別就是synchronous TCP/IP server。
# 這是一個糟糕的類設計，但是也儲存了一些設計的型別理念。
There are five classes in an inheritance diagram, four of which represent
synchronous servers of four types:
+------------+
| BaseServer |
+------------+
|
v
+-----------+ +------------------+
| TCPServer |------->| UnixStreamServer |
+-----------+ +------------------+
|
v
+-----------+ +--------------------+
| UDPServer |------->| UnixDatagramServer |
+-----------+ +--------------------+
# 下面是五個類的繼承關係圖表，其中的四個代表四種類型的同步伺服器：
+----------------+
| BaseServer |
+----------------+
|
v
+---------------+ +-------------------------+
| TCPServer |------->| UnixStreamServer |
+---------------+ +-------------------------+
|
v
+---------------+ +-----------------------------+
| UDPServer |------->| UnixDatagramServer |
+---------------+ +-----------------------------+
Note that UnixDatagramServer derives from UDPServer, not from
UnixStreamServer -- the only difference between an IP and a Unix
stream server is the address family, which is simply repeated in both
unix server classes.
# 注意：UnixDatagramServer繼承於UDPServer，而不是UnixStreamServer，
# IP和Unix stream server之間僅有的差異就是address family，兩個伺服器類的內容多數是簡單的重複。
Forking and threading versions of each type of server can be created
using the ForkingMixIn and ThreadingMixIn mix-in classes. For
instance, a threading UDP server class is created as follows:
class ThreadingUDPServer(ThreadingMixIn, UDPServer): pass
# forking和threading 可以被建立用於ForkingMixIn和TreadingMixIn mix-in類。
# 例如： threading UDP server類會被如下方式建立：
# class ThreadingUDPServer(ThreadingMixIn, UDPServer): pass
The Mix-in class must come first, since it overrides a method defined
in UDPServer! Setting the various member variables also changes
the behavior of the underlying server mechanism.
# Mix-in 這個類必須首先實現，因為它重寫了定義UDPServer的方法。
# 設定不同的成員變數也改變了基本的伺服器構造方法。
To implement a service, you must derive a class from
BaseRequestHandler and redefine its handle() method. You can then run
various versions of the service by combining one of the server classes
with your request handler class.
# 為了實現一個服務，你必須從基類BaseRequestHandler中重新定義它的handle方法。
# 然後通過把服務類與你重寫的Handle方法類結合，以此執行新的服務類
The request handler class must be different for datagram or stream
services. This can be hidden by using the request handler
subclasses StreamRequestHandler or DatagramRequestHandler.
# 請求處理類的TCP和UDP的方式是不同的，
# 這個可以通過使用請求處理的子類StreamRequestHandler或者DatagramRequestHandler來隱藏。
Of course, you still have to use your head!
For instance, it makes no sense to use a forking server if the service
contains state in memory that can be modified by requests (since the
modifications in the child process would never reach the initial state
kept in the parent process and passed to each child). In this case,
you can use a threading server, but you will probably have to use
locks to avoid two requests that come in nearly simultaneous to apply
conflicting changes to the server state.
# 當然，你還可以思考其他的方法。
# 例如，如果服務中包含請求修改的記憶體的狀態，那麼使用forking server沒有任何意義
# （因為在子程序中修改將不對父程序的初始化狀態有影響，父程序也不會把這個修改的引數傳遞給
# 其他子程序）。這種情況下，你可以使用threading server，而且你更有可能需要用到“鎖”，
# 以此來避免兩個請求同時到達而使伺服器狀態產生衝突。
On the other hand, if you are building e.g. an HTTP server, where all
data is stored externally (e.g. in the file system), a synchronous
class will essentially render the service "deaf" while one request is
being handled -- which may be for a very long time if a client is slow
to read all the data it has requested. Here a threading or forking
server is appropriate.
# 此外，如果你在搭建如HTTP伺服器等，所有的資料都會儲存在外部（如檔案系統中），
# 當客戶端的一項請求被處理時，並且客戶端的讀取資料的速度很慢，synchronous class將
# 會使服務不做出響應，這可能需要維持很長時間。
In some cases, it may be appropriate to process part of a request
synchronously, but to finish processing in a forked child depending on
the request data. This can be implemented by using a synchronous
server and doing an explicit fork in the request handler class
handle() method.
# 在一些情況下，請求同步可能需要恰當的方法，但是為了在子程序中完成請求要受到請求資料
# 的影響。這可以通過使用同步伺服器來實現，並且在請求處理類中的Handle方法中明確指定fork
# 的程序。
Another approach to handling multiple simultaneous requests in an
environment that supports neither threads nor fork (or where these are
too expensive or inappropriate for the service) is to maintain an
explicit table of partially finished requests and to use a selector to
decide which request to work on next (or whether to handle a new
incoming request). This is particularly important for stream services
where each client can potentially be connected for a long time (if
threads or subprocesses cannot be used).
# 另一種處理多個同時發生的請求的方法是維繫一張明確的完成請求的表單，使用
# select()方法來判定哪個請求應該在接下來做出響應（或者判斷是否要處理新到來的請求），
# 當每一個客戶端需要建立很長時間的連線時，這對於stream services來說非常重要。
# （前提是不使用執行緒和子程序的方法）
Future work:
- Standard classes for Sun RPC (which uses either UDP or TCP)
- Standard mix-in classes to implement various authentication
and encryption schemes
#未來工作：
# -Sun RPC的標準類(使用UDP或TCP)
# -用於實現各種身份驗證的標準混合類和加密方案
XXX Open problems:
- What to do with out-of-band data?
#開放問題：如何處理帶外資料？
BaseServer:
- split generic "request" functionality out into BaseServer class.
Copyright (C) 2000 Luke Kenneth Casson Leighton <[email protected]>
example: read entries from a SQL database (requires overriding
get_request() to return a table entry from the database).
entry is processed by a RequestHandlerClass.
#BaseServer:
# -將泛型“請求”功能分離到BaseServer類中。
# 版權所有(C) 2000 Luke Kenneth Casson Leighton
# 示例:從SQL資料庫中讀取條目(需要重寫)
# get_request()以從資料庫返回表項。
# 條目由RequestHandlerClass處理。
"""
# Author of the BaseServer patch: Luke Kenneth Casson Leighton
__version__ = "0.4"
import socket
import selectors
import os
import errno
import sys
try:
import threading
except ImportError:
import dummy_threading as threading
from io import BufferedIOBase
from time import monotonic as time
__all__ = ["BaseServer", "TCPServer", "UDPServer",
"ThreadingUDPServer", "ThreadingTCPServer",
"BaseRequestHandler", "StreamRequestHandler",
"DatagramRequestHandler", "ThreadingMixIn"]
if hasattr(os, "fork"):
__all__.extend(["ForkingUDPServer", "ForkingTCPServer", "ForkingMixIn"])
if hasattr(socket, "AF_UNIX"):
__all__.extend(["UnixStreamServer", "UnixDatagramServer",
"ThreadingUnixStreamServer",
"ThreadingUnixDatagramServer"])
# poll/select have the advantage of not requiring any extra file descriptor,
# contrarily to epoll/kqueue (also, they require a single syscall).
if hasattr(selectors, 'PollSelector'):
_ServerSelector = selectors.PollSelector
else:
_ServerSelector = selectors.SelectSelector
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.
# 每次處理一個請求，直到關閉。為每一個poll_interval秒進行關閉輪詢。
# 忽略了self.timeout。如果你需要做週期性的任務，那就做另一個執行緒。
"""
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.
# 考慮使用另一個檔案描述符或連線到
# 套接字來喚醒這個，而不是輪詢。輪詢會降低我們的
# 對關閉請求的響應性,會浪費cpu很多次。
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.
# 停止serve_forever迴圈。阻塞直到迴圈結束。這必須在while呼叫serve_forever()
# 正在另一個執行緒中執行，否則它會執行死鎖。
"""
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.
# 由serve_forever()迴圈呼叫。可能被子類/ Mixin覆蓋以實現任何程式碼需要在迴圈中執行。
# 由serve_forever()迴圈呼叫。可能被子類/ Mixin覆蓋以實現任何程式碼需要在迴圈中執行。
"""
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(),
# 操作請求函式是一個高級別的呼叫,它呼叫了selector.select().
# get_request(), verify_request() and process_request()
# 獲取請求函式\驗證函式和程序執行緒函式
# - get_request() is different for stream or datagram sockets
# get_request()對於流(TCP)或資料報(UDP)套接字是不同的
# - process_request() is the place that may fork a new process or create a
# new thread to finish the request
# process_request()可以派生一個新程序或建立一個新執行緒完成請求
# - finish_request() instantiates the request handler class; this
# constructor will handle the request all by itself
# finish_request()例項化請求處理程式類;這建構函式將自己處理請求
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.
# 支援使用socket.settimeout()，如果沒有將設定預設的
# self.timeout以確保可用
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().
#我假設select .select()返回的是套接字在呼叫此函式之前可讀，
因此應該在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)
def handle_timeout(self):
"""Called if no new request arrives within self.timeout.
# 如果self.timeout中沒有新請求到達，則呼叫。
Overridden by ForkingMixIn.
被ForkingMixIn類覆蓋
"""
pass
def verify_request(self, request, client_address):
"""Verify the request. May be overridden.
# 驗證這個請求,可能被覆蓋
Return True if we should proceed with this request.
# 如果我們繼續這個請求，返回True。
"""
return True
def process_request(self, request, client_address):
"""Call finish_request.
# 呼叫 finish_request
Overridden by ForkingMixIn and ThreadingMixIn.
# 被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."""
# 通過例項化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, file=sys.stderr)
print('Exception happened during processing of request from',
client_address, file=sys.stderr)
import traceback
traceback.print_exc()
print('-' * 40, file=sys.stderr)
def __enter__(self):
return self
def __exit__(self, *args):
self.server_close()
class TCPServer(BaseServer):
"""Base class for various socket-based server classes.
# 用於各種基於套接字的伺服器類的基類
Defaults to synchronous IP stream (i.e., TCP).
#預設為同步IP流(即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.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 selector.
# 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
# 明確地關閉。socket.close()僅僅是釋放
# the socket and waits for GC to perform the actual close.
# 套接字並等待GC執行實際的關閉。
request.shutdown(socket.SHUT_WR)
except OSError:
pass # some platforms may raise ENOTCONN here #一些平臺可能會在這裡引發ENOTCONN錯誤
self.close_request(request)
def close_request(self, request):
"""Called to clean up an individual request."""
# 呼叫以清理單個的請求
request.close()
class UDPServer(TCPServer):
"""UDP server class."""
# UDP服務類
allow_reuse_address = False
socket_type = socket.SOCK_DGRAM
max_packet_size = 8192
def get_request(self):
data, client_addr = self.socket.recvfrom(self.max_packet_size)
return (data, self.socket), client_addr
def server_activate(self):
# No need to call listen() for UDP.
# UDP服務類不需要listen()
pass
def shutdown_request(self, request):
# No need to shutdown anything.
# UDP服務類不需要
self.close_request(request)
def close_request(self, request):
# No need to close anything.
# UDP服務類不需要
pass
if hasattr(os, "fork"):
class ForkingMixIn:
"""Mix-in class to handle each request in a new process."""
# 混合類在新程序中處理每個請求
timeout = 300
active_children = None
max_children = 40
def collect_children(self):
"""Internal routine to wait for children that have exited."""
# 等待子程序離開的內部程式
if self.active_children is None:
return
# If we're above the max number of children, wait and reap them until
# we go back below threshold. Note that we use waitpid(-1) below to be
# able to collect children in size(<defunct children>) syscalls instead
# of size(<children>): the downside is that this might reap children
# which we didn't spawn, which is why we only resort to this when we're
# above max_children.
# 如果我們超過了子程序的最大數量，那需要等待並回收他們直到我們回到閾值以下。
# 注意，我們使用下面的waitpid(-1)來回收(<殭屍程序>)系統呼叫代替了回收子程序
# 缺點是這可能會沒有產生子程序，這就是為什麼我們只有在最大子程序數量之上
# 才會使用它。
while len(self.active_children) >= self.max_children:
try:
pid, _ = os.waitpid(-1, 0)
self.active_children.discard(pid)
except ChildProcessError:
# we don't have any children, we're done
# 沒有任何子程序
self.active_children.clear()
except OSError:
break
# Now reap all defunct children.
# 現在回收所有的殭屍程序
for pid in self.active_children.copy():
try:
pid, _ = os.waitpid(pid, os.WNOHANG)
# if the child hasn't exited yet, pid will be 0 and ignored by
# 如果這個子程序沒退出，PID會是0並被忽略
# discard() below
# 丟棄pid
self.active_children.discard(pid)
except ChildProcessError:
# someone else reaped it
# 其他的被回收了
self.active_children.discard(pid)
except OSError:
pass
def handle_timeout(self):
"""Wait for zombies after self.timeout seconds of inactivity.
等著不活動超時秒殭屍程序出現
May be extended, do not override.
可能被擴充套件，不要被覆蓋
"""
self.collect_children()
def service_actions(self):
"""Collect the zombie child processes regularly in the ForkingMixIn.
定期在ForkingMixin中收集殭屍子程序
service_actions is called in the BaseServer's serve_forver loop.
service_actions在BaseServer的serve_forver迴圈中被呼叫。
"""
self.collect_children()
def process_request(self, request, client_address):
"""Fork a new subprocess to process the request."""
# 派生一個新的子程序來處理請求
pid = os.fork()
if pid:
# Parent process
# 父程序
if self.active_children is None:
self.active_children = set()
self.active_children.add(pid)
self.close_request(request)
return
else:
# Child process.
# 子程序
# This must never return, hence os._exit()!
# 這裡必須永遠不能return,因此os._exit()!
status = 1
try:
self.finish_request(request, client_address)
status = 0
except Exception:
self.handle_error(request, client_address)
finally:
try:
self.shutdown_request(request)
finally:
os._exit(status)
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.
# 和在BaseServer中一樣，但作為一個執行緒。
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()
if hasattr(os, "fork"):
class ForkingUDPServer(ForkingMixIn, UDPServer): pass
class ForkingTCPServer(ForkingMixIn, TCPServer): pass
class ThreadingUDPServer(ThreadingMixIn, UDPServer): pass
class ThreadingTCPServer(ThreadingMixIn, TCPServer): pass
if hasattr(socket, 'AF_UNIX'):
class UnixStreamServer(TCPServer):
address_family = socket.AF_UNIX
class UnixDatagramServer(UDPServer):
address_family = socket.AF_UNIX
class ThreadingUnixStreamServer(ThreadingMixIn, UnixStreamServer): pass
class ThreadingUnixDatagramServer(ThreadingMixIn, UnixDatagramServer): 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.
這個類是為每個要處理的請求例項化的。建構函式設定例項化變數request client_address，
然後呼叫handle()方法，實現一個具體的服務，
您需要做的就是派生一個類定義handle()方法。
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.
# handle()方法可以將request查詢為self.request，客戶端地址為self.client_address，
# 以及伺服器(以防萬一)需要訪問每個伺服器的資訊)作為self.server。
# 每個請求都建立單獨的例項物件，handle()方法可以定義其他任意例項變數。
"""
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
# The following two classes make it possible to use the same service
# class for stream or datagram servers.
# 以下兩個類使使用相同的服務成為可能
# 用於流伺服器(TCP)或資料報伺服器(UDP)。
# Each class sets up these instance variables:
# - rfile: a file object from which receives the request is read
# - wfile: a file object to which the reply is written
# When the handle() method returns, wfile is flushed properly
# 每個類都設定這些例項變數:
# - rfile:接收請求的檔案物件
# - wfile:寫回復的檔案物件
# 當handle()方法返回時，wfile被適時地重新整理
class StreamRequestHandler(BaseRequestHandler):
"""Define self.rfile and self.wfile for stream sockets."""
# 為流伺服器定義self.rfile和self.wfile
# Default buffer sizes for rfile, wfile.
# We default rfile to buffered because otherwise it could be
# really slow for large data (a getc() call per byte); we make
# wfile unbuffered because (a) often after a write() we want to
# read and we need to flush the line; (b) big writes to unbuffered
# files are typically optimized by stdio even when big reads
# aren't.
# 預設的rfile, wfile 緩衝區大小
#我們預設將rfile設定為快取，否則可能會這樣對於大資料
# (每個位元組一個 getc()呼叫)來說確實很慢;
#我們做wfile未快取，因為(a)通常在寫入()之後閱讀，我們需要不斷重新整理;
#標準輸入輸出通常會對未緩衝的大檔案寫入操作進行優化，即使是大檔案的讀操作也是如此。
rbufsize = -1
wbufsize = 0
# A timeout to apply to the request socket, if not None.
# 應用於請求套接字(如果不是None)的超時。
timeout = None
# Disable nagle algorithm for this socket, if True.
# Use only when wbufsize != 0, to avoid small packets.
# 如果為這個套接字，禁用nagle演算法。
#僅在wbufsize != 0時使用，以避免小資料包。
disable_nagle_algorithm = False
def setup(self):
self.connection = self.request
if self.timeout is not None:
self.connection.settimeout(self.timeout)
if self.disable_nagle_algorithm:
self.connection.setsockopt(socket.IPPROTO_TCP,
socket.TCP_NODELAY, True)
self.rfile = self.connection.makefile('rb', self.rbufsize)
if self.wbufsize == 0:
self.wfile = _SocketWriter(self.connection)
else:
self.wfile = self.connection.makefile('wb', self.wbufsize)
def finish(self):
if not self.wfile.closed:
try:
self.wfile.flush()
except socket.error:
# A final socket error may have occurred here, such as
# the local error ECONNABORTED.
# 最後一個套接字錯誤可能發生在這裡，例如本地錯誤被終止。
pass
self.wfile.close()
self.rfile.close()
class _SocketWriter(BufferedIOBase):
"""Simple writable BufferedIOBase implementation for a socket
Does not hold data in a buffer, avoiding any need to call flush()."""
# 一個套接字的簡單可寫的BufferedIOBase實現
# 不將資料儲存在緩衝區中，避免呼叫flush()
def __init__(self, sock):
self._sock = sock
def writable(self):
return True
def write(self, b):
self._sock.sendall(b)
with memoryview(b) as view:
return view.nbytes
def fileno(self):
return self._sock.fileno()
class DatagramRequestHandler(BaseRequestHandler):
"""Define self.rfile and self.wfile for datagram sockets."""
# 定義self.rfile和self.wfile用於資料報套接字
def setup(self):
from io import BytesIO
self.packet, self.socket = self.request
self.rfile = BytesIO(self.packet)
self.wfile = BytesIO()
def finish(self):
self.socket.sendto(self.wfile.getvalue(), self.client_address)
先複習繼承知識
例子1
class Base:
def __init__(self,name):
self.name=name
def func(self):
print("from base")
class Son(Base):
def func(self):
print("from son")
sonobj= Son("nick")
sonobj.func()
輸出結果
from son
分析下執行過程
1、載入class Base ,載入class Son
2、執行sonobj= Son("nick")語句，生成sonobj物件，這裡首先尋找__init__初始化方法,Son類沒有，找Son的父類Base，找到父類的__init__方法，執行父類的方法，將引數"nick"傳入，注意這裡的self其實是sonobj,即類Son的物件，而非Base的物件，執行self.name=name，為sonobj物件增加了name屬性，初始化過程結束。
3、執行sonobj.func()語句，執行物件sonobj的func方法，首先要在這個sonobj物件自身的方法下面找，發現有func方法就執行該方法，而不需要再到父類Base執行func方法。
例子2
class Base:
def func(self):
print("from base")
class Son(Base):
def __init__(self,name):
self.name=name
def func(self):
print("from son")
class Base2:
def func(self):
print("from base2")
class Grandson(Base2,Son):
pass
sonobj= Grandson("nick")
sonobj.func()
輸出結果
from base2
簡單分析執行結果
1、在python3中所有的類都是新式類，執行的繼承順序是廣度優先，所以類Grandson繼承的順序是
2、首先執行初始化方法，在base2中沒找到，然後在Son中找到了，執行該方法，增加了物件的name屬性，
3、執行func方法，根據繼承順序，先左後右，先在base2中找到了，就直接執行func方法，不再繼續尋找。程式結束。
socketserver原始碼分析tcp版本
服務端
import socketserver
class Myserver(socketserver.BaseRequestHandler): #
def handle(self):
print("伺服器開始運行了")
print(self.request) # 相當於socket裡的conn連結
print(self.client_address) # 相當於socket裡的addr，即客戶端的ip地址
while True:
try:
res = self.request.recv(1024)
print(res.decode("utf-8"))
msg = input(">>>").strip()
self.request.send(msg.encode("utf-8"))
except Exception as e:
print(e)
break
if __name__ == "__main__":
ip, port = "127.0.0.1", 8090
serverobj = socketserver.ThreadingTCPServer((ip, port), Myserver)
serverobj.serve_forever()
客戶端
import socket
ip, port = "127.0.0.1", 8090
buffer_size = 1024
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect((ip, port))
while True:
msg = input(">>>").strip()
if msg == "q": break
if not msg: continue
client.send(('客戶端A：' + msg).encode("utf-8"))
res = client.recv(buffer_size)
print(res.decode("utf-8"))
分析內部執行過程：
1、
載入socketserver模組，載入class Myserver,執行ip, port = "127.0.0.1", 8090語句
2、
執行
serverobj = socketserver.ThreadingTCPServer((ip, port), Myserver)
語句，尋找初始化方法，根據socketserver模組各個類繼承關係圖可知，ThreadingTCPServer繼承了ThreadingMixIn, TCPServer，而TCPServer又繼承了BaseServer。
2.1、
首先在左邊的ThreadingMixIn尋找初始化方法，沒有找到，再在TCPServer裡尋找，找到了__init__方法，執行該方法。
TCPServer下的__init__方法
這裡的引數server_address就是傳入的(ip, port),而RequestHandlerClass就是自己定義的類Myserver
2.1.1、
這裡首先執行BaseServer下的__init__方法
BaseServer下的__init__方法
執行該方法下面的4個語句，注意這裡的self是ThreadingTCPServer的物件，而非BaseServer的物件，因為這次呼叫是ThreadingTCPServer的物件發起的，即服務端中的serverobj物件。
為serverobj增加了4個屬性，將(ip,port)賦值為self.server_addressz屬性，自己定義的類 Myserver賦值為RequestHandlerClass屬性，執行threading.Event()賦值為self.__is_shut_down。
2.1.2、
執行
self.socket = socket.socket(self.address_family,self.socket_type)
語句
建立socket物件並將socket物件賦值給self.socket屬性
2.1.3、
執行
if bind_and_activate:
try:
self.server_bind()
self.server_activate()
except:
self.server_close()
raise
2.1.3.1、
執行self.server_bind()語句，注意這裡的self還是serverobj物件，所以需要重新開始尋找server_bind方法，從左邊的ThreadingMixIn尋找，沒有找到，在右邊的TCPServer裡尋找，找到了server_bind()方法
由於self.allow_reuse_address=False，不執行這裡的if語句
self.socket.bind(self.server_address)
執行了socket物件的繫結ip\port,
執行
self.server_address = self.socket.getsockname()
將返回套接字自己的地址。通常是一個元組(ipaddr,port)重新賦值給self.server_address 屬性
2.1.3.2、
執行self.server_activate()語句，與上面一樣，先從ThreadingMixIn找，再找TCPServer，執行TCPServer下面的server_activate方法，
self.socket.listen(self.request_queue_size)
執行socket物件的監聽動作
2.1.3.3、
如果出錯，執行self.server_close()，還是一樣的尋找過程，這裡就不寫了，直接寫找到的結果，執行
TCPServer下面的server_close()方法，執行self.socket.close()語句，即關閉socket
2.2、
至此，serverobj物件的初始化方法__init__才執行完
3、