Python網路編程中的伺服器架構（負載均衡、單線程、多線程和同步、非同步等）。

03-03

這篇文章主要介紹伺服器架構。

網路服務需要面對兩個挑戰。

第一個問題是核心挑戰，要編寫出能夠正確處理請求並構造合適響應的代碼。
第二個挑戰是如何將網路代碼部署到隨系統自動啟動的Windows服務或者是Unix守護進程中，將活動日誌持久化存儲。並且在無法連接到資料庫或者後端存儲區時發出警告，為其提供完整的保護，以防止所有可能的失敗情形，或是確保其在失敗時快速重啟。

這篇文章直重點說第一個問題。然後會介紹伺服器部署，然後把重點放在如何構建網路伺服器軟體上。

首先，我們可以很自然的把網路伺服器分為三大類。

第一類就是簡單的單線程伺服器（比如UDP伺服器和TCP伺服器），在這裡會詳細說明這類伺服器的局限性，即同一時刻只能為一個客戶端服務，此時其他客戶端只能等待。即使為一個客戶服務，這時候CPU也可能處於近乎空閑的狀態。

第二類就是解決局限性的一個方案，使用多個線程或者進程，每個線程或者進程內都運行一個單線程伺服器。

第三類就是與第二類剛好相對的另一種解決方案，在自己的代碼中使用非同步網路操作來支持多路復用，而不直接使用操作系統提供的多路復用。

然後我們說一下伺服器部署，我們可能會把網路服務部署到單台機器上，也可能部署到多台機器上。要使用單台機器上的服務，客戶端只要鏈接到該機器的IP地址即可，而要使用運行在多台機器上的服務，就需要更加複雜的方法。

一種方法是將這個服務的某個實例的地址或者主機名返回給客戶端（比如與客戶端運行在同一機房中服務實例），但是這種方法沒有提供冗餘性，如果服務的這一個實例宕機了，那麼通過主機名或者IP地址硬編碼鏈接這個服務實例的客戶端都無法繼續鏈接。

另外一種更加健壯性的方法就是，當要訪問某個服務時候，令DNS伺服器返回運行這個服務的所有IP地址，如果客戶端無法連接到第一個地址的話，可以連接到第二個地址，然後第三個。工業界一般會在服務前配置一個負載均衡器，客戶端直接連接到負載均衡器，然後由負載均衡器將鏈接請求轉發到實際的伺服器。如果某台伺服器宕機了，那麼負載均衡器會將轉發至該伺服器的連接請求予以停止。直到這個伺服器恢復服務為止。這樣伺服器的故障對於大量的用戶來說是不可見的。

大型的互聯網服務中結合了這兩個方法：每個機房中都配置了一個負載均衡器與伺服器群，而公共的DNS名會返回與用戶距離最近的機房中的負載均衡器的IP地址。

無論伺服器架構多麼簡單或者多麼複雜，都需要使用某種方式在物理或者虛擬機器上運行我們的Python伺服器代碼，這一個過程叫做部署。

對於部署來說，比較舊式的技術觀點就是，為每個伺服器程序都編寫服務所提供的所有功能：通過兩次fork()創建一個Unix守護進程（或者是將自己註冊為一個Windows服務），安排進行系統級的日誌操作，支持配置文件以及提供啟動、關閉和重啟的相關機制。可以使用已經解決了相關問題的第三方庫來完成伺服器程序的編寫，也可以在自己的代碼中重新實現這些功能。

另外一種方法就是。提倡只是先伺服器程序必備功能的最小集合。它將每個服務實現為普通的前台程序，而不是將其實現為守護進程。這樣的程序從環境變數（Python中的sys.environ字典）而不是系統級的配置文件中獲取所需要的配置選項。他通過環境變數中指定的選項鏈接到任意的後端服務，並且直接將日誌信息輸出到屏幕，甚至直接使用Python自己提供的print()函數。另外，這個方法通過打開並且監聽環境配置指定的任意埠來接受網路請求。

現在有一些大型的平台服務提供商提供了託管這種程序的功能。他們將應用程序的幾十個甚至幾百個副本配置在一個公共域名和TCP負載均衡器下，然後將所有輸出的日誌聚集起來進行分析。這些提供商允許我們直接提交代碼。但是更多的提供商，更希望我們提供代碼、Python解釋器以及所需要的依賴打包入一個容器內。（說到這裡，可能你會想起Docker）。我們可以在自己的筆記本電腦上對這個容器進行測試，然後將其部署到生產環境中，從而能夠確認，生產環境中運行的Python代碼與測試環境中運行的代碼使用的是完全相同的鏡像。無論是用哪種方法，都無需在單個服務中提供多個功能，服務中所有的冗餘和重複都可以讓平台來處理。

下面是一段示例代碼。這段代碼使用的是一個最簡單的TCP協議進行說明。在這個協議中，客戶端可以詢問3個問題，這三個問題都使用純文本的ASCII字元表示。

這三個問題都是基於The Zen of Python中的格言。可以通過import this來獲取這首詩。

為了基於這個協議構建一個客戶端和多個伺服器，這裡面定義了很多規則。這個代碼本身並沒有命令行介面。該程序編寫的模塊存在的唯一作用就是作為一個支持性模塊讓後續的代碼導入。接下來的代碼也可以用下面這段代碼中定義的模式，而不需要重複編寫。

import argparse, socket, timeaphorisms = {bBeautiful is better than?: bUgly., bExplicit is better than?: bImplicit., bSimple is better than?: bComplex.}def get_answer(aphorism): """Return the string response to a particular Zen-of-Python aphorism.""" time.sleep(0.0) # increase to simulate an expensive operation return aphorisms.get(aphorism, bError: unknown aphorism.)def parse_command_line(description): """Parse command line and return a socket address.""" parser = argparse.ArgumentParser(description=description) parser.add_argument(host, help=IP or hostname) parser.add_argument(-p, metavar=port, type=int, default=1060, help=TCP port (default 1060)) args = parser.parse_args() address = (args.host, args.p) return addressdef create_srv_socket(address): """Build and return a listening server socket.""" listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) listener.bind(address) listener.listen(64) print(Listening at {}.format(address)) return listenerdef accept_connections_forever(listener): """Forever answer incoming connections on a listening socket.""" while True: sock, address = listener.accept() print(Accepted connection from {}.format(address)) handle_conversation(sock, address)def handle_conversation(sock, address): """Converse with a client over `sock` until they are done talking.""" try: while True: handle_request(sock) except EOFError: print(Client socket to {} has closed.format(address)) except Exception as e: print(Client {} error: {}.format(address, e)) finally: sock.close()def handle_request(sock): """Receive a single client request on `sock` and send the answer.""" aphorism = recv_until(sock, b?) answer = get_answer(aphorism) sock.sendall(answer)def recv_until(sock, suffix): """Receive bytes over socket `sock` until we receive the `suffix`.""" message = sock.recv(4096) if not message: raise EOFError(socket closed) while not message.endswith(suffix): data = sock.recv(4096) if not data: raise IOError(received {!r} then socket closed.format(message)) message += data return message

其中最後的四個函數展示了伺服器進程的核心模式。這四個函數層級調用包括了監聽套接字來創建TCP伺服器的內容，以及關於數據封幀和錯誤處理的內容。

上面這段代碼就是用來構建各種伺服器的工具箱。

為了測試這篇文章中的伺服器，需要一個客戶端程序。下面這段代碼提供了一個簡單的命令行工具作為客戶端。

import argparse, random, socket, zen_utilsdef client(address, cause_error=False): sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.connect(address) aphorisms = list(zen_utils.aphorisms) if cause_error: sock.sendall(aphorisms[0][:-1]) return for aphorism in random.sample(aphorisms, 3): sock.sendall(aphorism) print(aphorism, zen_utils.recv_until(sock, b.)) sock.close()if __name__ == __main__: parser = argparse.ArgumentParser(description=Example client) parser.add_argument(host, help=IP or hostname) parser.add_argument(-e, action=store_true, help=cause an error) parser.add_argument(-p, metavar=port, type=int, default=1060, help=TCP port (default 1060)) args = parser.parse_args() address = (args.host, args.p) client(address, args.e)

在正常情況下，cause_error為False。此時客戶端將創建一個TCP套接字，然後發送3句格言作為請求，每發送一個就等待伺服器返回相應的答案。

如果想知道如何處理輸入有誤的情況，客戶端提供了-e的選項，用來發送不完整的問題，使伺服器突然掛起。如果伺服器已經啟動並且正確運行，就能在客戶端看到這三個問題以及相應的答案。

下面我們說下一個問題，單線程伺服器。上面第一段代碼中zen_utils模塊提供了豐富的工具程序。減少了很多編寫一個簡單的單線程伺服器的工作量。單線程伺服器是最簡單的可用設計。下面只用3行代碼就可以完成這個單線程伺服器。

import zen_utilsif __name__ == __main__: address = zen_utils.parse_command_line(simple single-threaded server) listener = zen_utils.create_srv_socket(address) zen_utils.accept_connections_forever(listener)

多餘的事情不再詳細說明了，直接說單線程伺服器的缺點。

如果伺服器和一個客戶端進行會話期間，另外一個客戶端也嘗試連接伺服器，這個設計就會出現問題了。只要伺服器與第一個客戶端的會話沒有完成，新建立的鏈接就會一直處於操作系統的監聽隊列中。

對單線程伺服器進行拒接服務攻擊是非常容易，連接該伺服器，並且永遠不關閉這個鏈接就可以了。

而且單線程設計無法在等待客戶端發送下一個請求時進行其他操作，因此很浪費伺服器的CPU和系統資源。

所以，下面會說如何克服單線程伺服器的這些限制。

我們說第一個方法。構建多線程與多進程伺服器。伺服器可以同時與多個客戶端進行會話，利用操作系統的內置支持，使用多個控制線程單獨運行同一段代碼。可以創建多個共享相同內存空間的線程，也可以創建完全獨立運行的線程。

這個方法的優點是：簡潔，直接使用單線程伺服器的代碼，創建多個線程運行它的副本。

這個方法的缺點是：伺服器能夠同時通信的客戶端數量收到操作系統並發機制規模的限制。即使某個客戶端處於空閑狀態，或者是緩慢運行狀態，他也會佔用整個線程或進程。就算程序被recv()阻塞,也會佔用系統RAM以及進程表中的一個進程槽。當同時運行的線程數量達到幾千甚至更多的時候，操作系統很少能夠維持良好的表現。此時系統在切換服務的客戶端時候，需要進行大量上下文切換，這使得運行效率大大降低。

每個線程都可以擁有伺服器監聽套接字的一個副本，並運行自己的accept()函數。操作系統會將每個新的客戶端鏈接交給任何運行了accept()函數並處於等待的線程來處理。如果所有線程都處在繁忙狀態的話，操作系統會將鏈接置於隊列中，直到每個線程為止。

import zen_utilsfrom threading import Threaddef start_threads(listener, workers=4): t = (listener,) for i in range(workers): Thread(target=zen_utils.accept_connections_forever, args=t).start()if __name__ == __main__: address = zen_utils.parse_command_line(multi-threaded server) listener = zen_utils.create_srv_socket(address) start_threads(listener)

這是多線程程序的一個可能設計。主線程啟動n個伺服器線程。然後退出。主線程認為這n個工作線程將永遠運行。因此運行這些線程的進程也會保持運行狀態。

當然也有其他的可選設計。比如主線程可以保持運行，並且成為一個伺服器線程。

上面介紹的是使用了操作系統級的控制線程來處理同一時刻的多個客戶端對話，而且Python標準庫也內置了一個框架socketsever，實現了這一個模式。

這個模塊將多線程模式分為了兩個：第一個是用於打開監聽套接字並接受客戶端鏈接的sever模式，第二個是用於通過某個打開的套接字與特定的客戶端進行繪畫的handler模式。結合這兩個模式，我們需要實例化一個sever對象，然後將一個handler對象作為參數傳給sever對象。下面是示例代碼：

from socketserver import BaseRequestHandler, TCPServer, ThreadingMixInimport zen_utilsclass ZenHandler(BaseRequestHandler): def handle(self): zen_utils.handle_conversation(self.request, self.client_address)class ZenServer(ThreadingMixIn, TCPServer): allow_reuse_address = 1 # address_family = socket.AF_INET6 # uncomment if you need IPv6if __name__ == __main__: address = zen_utils.parse_command_line(legacy "SocketServer" server) server = ZenServer(address, ZenHandler) server.serve_forever()

當然，我們也可以將ThreadingMixIn改為ForkingMixIn,這樣就可以使用完全隔離的進程來處理鏈接的客戶端，而不使用線程。

上述代碼的一個缺點就是，不限制伺服器最終啟動的線程數量，這樣很容易過載。

下面我們說另外一個解決方法，那就是非同步伺服器。如果想在不為每個客戶端分配一個操作系統級的控制線程的前提下保證CPU在這段時間內處於繁忙狀態。我們可以使用非同步模式來編寫伺服器。

使用這種模式代碼不需要等待數據發送至某一個特定的客戶端或由這個客戶端接收。相反，代碼可以從整個處於等待的客戶端套接字列表中讀取數據。

現代操作系統網路棧讓這個模式的應用成為了現實。網路棧提供了一個操作系統調用。支持進程為等待整個客戶端套接字列表中的套接字而阻塞。而不只是等待一個單獨的客戶端套接字。另外一個特點是，可以將一個套接字配置為非阻塞套接字。

非同步這個屬於表示伺服器代碼從來不會停下來等待某一個特定的客戶端，即代碼的控制線程不是同步的。非同步伺服器可以在所有連接的客戶端之前自由切換，並提供相應的服務。

操作系統也有很多調用是支持非同步的代碼。最古老的就是POSIX的select()調用。不過調用很多方面效率低下。

現在操作系統出現了一些select()的替代品，比如Linux上面的poll()，和BSD系統上的epoll()調用。

下面是一個簡單非同步伺服器的完整內部細節。

import select, zen_utilsdef all_events_forever(poll_object): while True: for fd, event in poll_object.poll(): yield fd, eventdef serve(listener): sockets = {listener.fileno(): listener} addresses = {} bytes_received = {} bytes_to_send = {} poll_object = select.poll() poll_object.register(listener, select.POLLIN) for fd, event in all_events_forever(poll_object): sock = sockets[fd] # Socket closed: remove it from our data structures. if event & (select.POLLHUP | select.POLLERR | select.POLLNVAL): address = addresses.pop(sock) rb = bytes_received.pop(sock, b) sb = bytes_to_send.pop(sock, b) if rb: print(Client {} sent {} but then closed.format(address, rb)) elif sb: print(Client {} closed before we sent {}.format(address, sb)) else: print(Client {} closed socket normally.format(address)) poll_object.unregister(fd) del sockets[fd] # New socket: add it to our data structures. elif sock is listener: sock, address = sock.accept() print(Accepted connection from {}.format(address)) sock.setblocking(False) # force socket.timeout if we blunder sockets[sock.fileno()] = sock addresses[sock] = address poll_object.register(sock, select.POLLIN) # Incoming data: keep receiving until we see the suffix. elif event & select.POLLIN: more_data = sock.recv(4096) if not more_data: # end-of-file sock.close() # next poll() will POLLNVAL, and thus clean up continue data = bytes_received.pop(sock, b) + more_data if data.endswith(b?): bytes_to_send[sock] = zen_utils.get_answer(data) poll_object.modify(sock, select.POLLOUT) else: bytes_received[sock] = data # Socket ready to send: keep sending until all bytes are delivered. elif event & select.POLLOUT: data = bytes_to_send.pop(sock) n = sock.send(data) if n < len(data): bytes_to_send[sock] = data[n:] else: poll_object.modify(sock, select.POLLIN)if __name__ == __main__: address = zen_utils.parse_command_line(low-level async server) listener = zen_utils.create_srv_socket(address) serve(listener)

這段事件循環代碼的精髓在於，它使用了自己的數據結構來維護每個客戶端會話的狀態，而沒有依賴操作系統在客戶端活動改變時進行上下文切換。伺服器有兩層循環。首先是一個不斷調用poll()的while循環。一次poll()調用可能返回多個事件，因此這個while循環內部還有一個循環，用於處理poll()返回的每一個事件。我們將這兩層迭代隱藏在一個生成器內，這樣就避免了主伺服器循環因為這兩次循環迭代而多用兩個不必要的縮進。

然後程序維護了sockets字典。從poll()獲取表示已經準備好進行後續通信對的套接字文件描述符n後，就能夠根據該文件描述符從sockets字典中查找到相應的Python套接字了。我們還在裡面儲存了套接字的地址。這樣，即使套接字關閉，操作系統也無法繼續提供已經連接好的地址。也能夠列印出正確的遠程地址作為調試信息。

這個伺服器的真正核心其實是它的緩衝區：在等待某個請求完成時，會將收到的數據存儲在bytes_received字典中。在等待操作系統安排發送數據時，會將要發送的位元組存儲在bytes_to_send字典中。這兩個緩衝區與我們告知poll()要在每個套接字上等待的時間一起形成了一個完整的狀態機。用於一步一步的處理客戶端會話。

準備連接的客戶端首先會將它自身視作伺服器監聽套接字上的一個事件，要始終將該事件設置為POLLIN（poll input）狀態。
當套接字本身就是客戶端套接字，並且事件類型為POLLIN時。就能夠使用recv()方法接收到最多為4KB的數據了。
套接字設置為POLLOUT後，只要客戶端套接字的發送緩衝區還能夠接收一個或者多個位元組，那麼poll()的調用就會立刻通知我們。
最後，如果套接字模式設置為POLLOUT後，並且send()完成了所有的數據發送，那麼此時就完成了一個完整的請求-響應循環，因此將套接字模式切換為POLLIN，用於下一個請求。
如果客戶端套接字返回了錯誤信息或者是錯誤狀態，就將這個客戶端的套接字以及發送緩衝區與接受緩衝區丟棄。

這個非同步方法的關鍵之處在於，可以在一個控制線程中處理成千上萬的客戶端會話。當每個客戶套接字準備好下一個事件時，代碼就執行該套接字的下一個操作，接收或發送數據。然後立刻返回到poll()調用，監控更多事件。

如果香江注意力放在客戶端代碼上，而將與select()、poll()或是epoll()有關的細節交給別人去負責。就可以看一下下面說的。下面是兩種風格。

第一種是回調風格的asyncio：

asynic框架支持兩種編程風格。第一種風格就是使用Twisted框架時，用戶通過對象實例來維護每個打開的客戶端連接。在這種設計模式中，使用對象實例上的方法調用代替了上述代碼中用來加速客戶端會話的各步驟。

下面仍舊是讀取問題，然後給出響應。回調風格的asyncio：

import asyncio, zen_utilsclass ZenServer(asyncio.Protocol): def connection_made(self, transport): self.transport = transport self.address = transport.get_extra_info(peername) self.data = b print(Accepted connection from {}.format(self.address)) def data_received(self, data): self.data += data if self.data.endswith(b?): answer = zen_utils.get_answer(self.data) self.transport.write(answer) self.data = b def connection_lost(self, exc): if exc: print(Client {} error: {}.format(self.address, exc)) elif self.data: print(Client {} sent {} but then closed .format(self.address, self.data)) else: print(Client {} closed socket.format(self.address))if __name__ == __main__: address = zen_utils.parse_command_line(asyncio server using callbacks) loop = asyncio.get_event_loop() coro = loop.create_server(ZenServer, *address) server = loop.run_until_complete(coro) print(Listening at {}.format(address)) try: loop.run_forever() finally: server.close() loop.close()

可以通過該框架來獲取遠程地址，而不是直接通過套接字來獲取。數據是通過一個方法調用來傳輸。這個方法只需要將接收到的字元串作為參數。

第二種風格是，協程風格的asyncio：

asyncio框架提供的另外一種構造協議代碼的方法就是使用協程。協程是一個函數，它在進行I/O操作時不會阻塞，而是會暫停，並將控制權轉移回調用方。Python語言的支持協程的一種標準形式就是生成器——在內部包含一個或多個yield語句的函數。這類函數不會再運行了一條返回語句之後就退出，而是會返回一個序列。

下面是通過協程實現的Zen協議。

import asyncio, zen_utils@asyncio.coroutinedef handle_conversation(reader, writer): address = writer.get_extra_info(peername) print(Accepted connection from {}.format(address)) while True: data = b while not data.endswith(b?): more_data = yield from reader.read(4096) if not more_data: if data: print(Client {} sent {!r} but then closed .format(address, data)) else: print(Client {} closed socket normally.format(address)) return data += more_data answer = zen_utils.get_answer(data) writer.write(answer)if __name__ == __main__: address = zen_utils.parse_command_line(asyncio server using coroutine) loop = asyncio.get_event_loop() coro = asyncio.start_server(handle_conversation, *address) server = loop.run_until_complete(coro) print(Listening at {}.format(address)) try: loop.run_forever() finally: server.close() loop.close()

上面介紹的非同步伺服器都可以在服務的不同客戶端會話間切換。要完成切換，只需要掃描協議對象即可。

然而，非同步伺服器是有硬性限制的。因為所有的操作都在單個操作系統線程中完成。所以一旦CPU使用率達到么100%，非同步伺服器就無法再為任何客戶端提供服務。這個時候即使有多個核心，所有工作也只能在單個處理器上完成。

然後這裡有一個兩全其美的方法。當我們需要高性能的時候，我們首先使用非同步對象或協程來編寫服務，並通過非同步框架來啟動服務。然後再回過頭來配置一些運行伺服器的操作系統，檢查操作系統CPU內核數目。有多少個CPU內核，就啟動多少個事件循環。

下面說一下inted守護進程。他解決了下面的問題：在一台特定的伺服器上，在系統啟動時啟動n個不同的後台進程，用於提供n個不同的網路服務。可以簡單地在系統的/etc/inted.conf文件中將所有要監聽的埠全部列出。

inted守護進程在列出的每個埠都調用了bind()和listen()，不過它只在客戶端真正連接時才會啟動一個伺服器進程。

為每一個連接都建立一個進程的花銷是很大的，而且會降低伺服器的利用率，不過這種方法也更加簡單，要通過這種方式啟動服務，只要在該服務的inetd.conf配置文件中將第4個欄位設置為nowait即可。

如果是下面的命令：

>>>1060 stream tcp nowait brandon /user/bin/python3 /user/bin/python3 in_zen1.py

這樣的服務一經啟用，其標準輸入輸出流、標準輸出流以及標準錯誤流便被連接到客戶端套接字。服務只需要與連接的客戶端通信，然後退出即可。

下面是和上面的inetd.conf配置結合使用的例子：

import socket, sys, zen_utilsif __name__ == __main__: sock = socket.fromfd(0, socket.AF_INET, socket.SOCK_STREAM) sys.stdin = open(/dev/null, r) sys.stdout = sys.stderr = open(/tmp/zen.log, a, buffering=1) address = sock.getpeername() print(Accepted connection from {}.format(address)) zen_utils.handle_conversation(sock, address)

另一種模式是，在配置inetd.conf時將第四個欄位指定為wait，表示會監聽套接字提供給腳本。腳本需要調用accept()，勇於接受正在等待的客戶端的連接請求。

這一個模式優勢在於，伺服器可以保持運行狀態。並不斷運行accept()來接受更多的客戶端的連接請求，而在這一過程中並需要inetd的介入。如果客戶端暫時停止了連接，伺服器也可以自由調用exit()，來降低伺服器的內存佔用。在客戶端再次需要伺服器的時候再啟動伺服器即可。inetd會檢測到我們的服務已經退出，然後會由inetd來負責監聽。

下面是wait模式設計的代碼。

import socket, sys, zen_utilsif __name__ == __main__: listener = socket.fromfd(0, socket.AF_INET, socket.SOCK_STREAM) sys.stdin = open(/dev/null, r) sys.stdout = sys.stderr = open(/tmp/zen.log, a, buffering=1) listener.settimeout(8.0) try: zen_utils.accept_connections_forever(listener) except socket.timeout: print(Waited 8 seconds with no further connections; shutting down)

有的inetd版本還內置了一種基於IP地址與主機名的簡單訪問控制機制。有興趣的可以查看相關的資料。

這篇文章到了這裡就結束了，將編寫的伺服器安裝在伺服器上，並且在系統啟動時運行伺服器的過程叫做部署。

以後的文章會講一些Python程序員依賴的基礎網路服務，HTTP協議的設計還會講到一些Python工具。還會說到Tornado的非同步框架。

不過，那都是後話了。

新的一年天天開心。所有的一切萬事勝意。