返回文章列表
玄貓（BlackCat）

Asyncio 非同步網路伺服器與客戶端實作

本文深入探討如何使用 Python 的 Asyncio 函式庫建立非同步網路伺服器和客戶端。文章涵蓋 TCP 伺服器和客戶端的程式碼範例，並示範如何使用 aiohttp 簡化 Web 伺服器開發。此外，文章還提供伺服器效能統計方法，以及連線建立和資料接收的程式碼邏輯，並輔以流程圖解說，幫助讀者理解 Asyncio

網路程式設計 Python
Asyncio TCP 伺服器 客戶端 aiohttp 網路

Asyncio 作為 Python 非同步程式設計的核心，為網路應用開發提供了高效能的解決方案。本文從基礎的 TCP 伺服器和客戶端實作開始，逐步引導讀者理解 Asyncio 的核心概念。接著，我們將介紹如何利用 aiohttp 框架簡化 Web 伺服器的開發流程，並提供效能統計的實作方法，讓讀者能夠監控伺服器的執行狀態。最後，我們將深入探討伺服器連線建立和資料接收的底層邏輯，並以流程圖的方式清晰地呈現資料的流動過程，幫助讀者更好地掌握 Asyncio 的應用技巧。

非同步網路伺服器

asyncio 提供了一個實用的框架來實作非同步網路伺服器。以下是一個簡單的 TCP 伺服器範例，示範如何使用 asyncio 來建立一個非同步網路伺服器。

TCP 伺服器範例

import asyncio

# 伺服器位址
SERVER_ADDRESS = ('0.0.0.0', 1234)

class YellEchoServer(asyncio.Protocol):
    def connection_made(self, transport):
        # 儲存傳輸物件
        self.transport = transport
        # 列印連線資訊
        print("Connection received from:", transport.get_extra_info('peername'))

    # 處理接收到的資料
    def data_received(self, data):
        # 將接收到的資料轉換為字串
        message = data.decode()
        # 列印接收到的資料
        print("Received:", message)
        # 將資料轉換為大寫並發送回客戶端
        response = message.upper().encode()
        self.transport.write(response)

# 建立事件迴圈
loop = asyncio.get_event_loop()

# 建立伺服器
factory = lambda: YellEchoServer()
coro = loop.create_server(factory, *SERVER_ADDRESS)

# 啟動伺服器
server = loop.run_until_complete(coro)

# 列印伺服器啟動資訊
print("Server started. Listening on", SERVER_ADDRESS)

# 執行事件迴圈
try:
    loop.run_forever()
except KeyboardInterrupt:
    pass

# 關閉伺服器
server.close()
loop.run_until_complete(server.wait_closed())
loop.close()

使用 aiohttp 建立 Web 伺服器

雖然 asyncio 可以用於建立非同步 Web 伺服器，但是在大多數情況下，使用 aiohttp 會更方便和高效。aiohttp 是一個根據 asyncio 的非同步 Web 框架，提供了簡單易用的 API 來建立 Web 伺服器。

aiohttp 範例

from aiohttp import web

async def handle_request(request):
    # 處理請求
    return web.Response(text="Hello, World!")

app = web.Application()
app.add_routes([web.get('/', handle_request)])

if __name__ == '__main__':
    web.run_app(app)

使用Asyncio實作TCP伺服器和客戶端

伺服器實作

要實作一個TCP伺服器，首先需要定義一個繼承自asyncio.Protocol的類別。這個類別將會定義如何處理客戶端的連線和資料傳輸。

import asyncio

class YellEchoServer(asyncio.Protocol):
    def connection_made(self, transport):
        # 連線建立時呼叫
        self.transport = transport

    def data_received(self, data):
        # 收到資料時呼叫
        self.transport.write(data.upper())

    def connection_lost(self, exc):
        # 連線斷開時呼叫
        print("Client disconnected")

# 建立事件迴圈
event_loop = asyncio.get_event_loop()

# 建立伺服器工廠
factory = event_loop.create_server(YellEchoServer, 'localhost', 1234)

# 啟動伺服器
server = event_loop.run_until_complete(factory)

try:
    # 執行事件迴圈
    event_loop.run_forever()
finally:
    # 關閉伺服器
    server.close()
    event_loop.run_until_complete(server.wait_closed())
    event_loop.close()

客戶端實作

要實作一個TCP客戶端，同樣需要定義一個繼承自asyncio.Protocol的類別。

import asyncio

class EchoClientProtocol(asyncio.Protocol):
    def __init__(self, message, loop):
        self.message = message
        self.loop = loop

    def connection_made(self, transport):
        # 連線建立時呼叫
        transport.write(self.message.encode())

    def data_received(self, data):
        # 收到資料時呼叫
        print(data.decode())
        self.loop.stop()

    def connection_lost(self, exc):
        # 連線斷開時呼叫
        print("Server disconnected")
        self.loop.stop()

# 建立事件迴圈
loop = asyncio.get_event_loop()

# 建立客戶端工廠
message = "Hello, World!".encode()
coro = loop.create_connection(lambda: EchoClientProtocol(message, loop), 'localhost', 1234)

# 啟動客戶端
loop.run_until_complete(coro)
loop.run_forever()

測試

可以使用nc命令來測試伺服器和客戶端。

$ nc localhost 1234
hello world!
HELLO WORLD!
^D

輸入任何文字後按下Enter鍵，伺服器將會回傳大寫的文字。按下Ctrl+D鍵可以關閉連線。

使用Asyncio建立TCP客戶端和伺服器

在這個例子中，我們將使用Python的Asyncio函式庫建立一個TCP客戶端和伺服器。客戶端將向伺服器傳送訊息，伺服器則會將收到的訊息轉換為大寫後發回給客戶端。

客戶端程式碼

import asyncio

class EchoClientProtocol(asyncio.Protocol):
    def __init__(self, message, loop):
        self.message = message
        self.loop = loop

    def connection_made(self, transport):
        self.transport = transport
        self.talk()

    def talk(self):
        self.transport.write(self.message)

    def data_received(self, data):
        self.talk()

    def connection_lost(self, exc):
        self.loop.stop()

loop = asyncio.get_event_loop()
loop.run_until_complete(loop.create_connection(
    lambda: EchoClientProtocol(b'Hello World!', loop),
    '127.0.0.1', 1234))
try:
    loop.run_forever()
finally:
    loop.close()

伺服器程式碼

import asyncio
import time

SERVER_ADDRESS = ('0.0.0.0', 1234)

class YellEchoServer(asyncio.Protocol):
    def __init__(self, stats):
        self.stats = stats
        self.stats['started at'] = time.time()

    def connection_made(self, transport):
        self.transport = transport

    def data_received(self, data):
        self.transport.write(data.upper())

    def connection_lost(self, exc):
        pass

stats = {}
loop = asyncio.get_event_loop()
server = loop.run_until_complete(loop.create_server(
    lambda: YellEchoServer(stats),
    *SERVER_ADDRESS))
try:
    loop.run_forever()
finally:
    server.close()
    loop.close()

統計伺服器效能

為了測試伺服器的效能，我們可以修改伺服器程式碼以收集統計資料。例如，我們可以計算伺服器處理的請求數量、處理時間等。

import asyncio
import time

SERVER_ADDRESS = ('0.0.0.0', 1234)

class YellEchoServer(asyncio.Protocol):
    def __init__(self, stats):
        self.stats = stats
        self.stats['started at'] = time.time()
        self.stats['requests'] = 0

    def connection_made(self, transport):
        self.transport = transport

    def data_received(self, data):
        self.stats['requests'] += 1
        self.transport.write(data.upper())

    def connection_lost(self, exc):
        pass

stats = {}
loop = asyncio.get_event_loop()
server = loop.run_until_complete(loop.create_server(
    lambda: YellEchoServer(stats),
    *SERVER_ADDRESS))
try:
    loop.run_forever()
finally:
    server.close()
    loop.close()
    print('伺服器統計資料:')
    print('啟動時間:', stats['started at'])
    print('請求數量:', stats['requests'])

這樣，我們就可以使用Asyncio建立TCP客戶端和伺服器，並測試伺服器的效能。

建立非同步伺服器的連線與資料接收

在建立非同步伺服器時，需要定義連線建立和資料接收的邏輯。以下是相關的程式碼實作：

連線建立

def connection_made(self, transport):
    """
    當連線建立時，更新傳輸物件和連線資料。
    """
    self.transport = transport
    self.stats['connections'] += 1

資料接收

def data_received(self, data):
    """
    當接收到資料時，將其轉換為大寫並發送回給客戶端。
    """
    self.transport.write(data.upper())
    self.stats['messages sent'] += 1

啟動伺服器

event_loop = asyncio.get_event_loop()

stats = {
    "connections": 0,
    "messages sent": 0,
    "started at": time.time()  # 記錄伺服器啟動時間
}

factory = event_loop.create_server(
    lambda: YellEchoServer(stats), *SERVER_ADDRESS
)

server = event_loop.run_until_complete(factory)
try:
    event_loop.run_forever()
except KeyboardInterrupt:
    pass
finally:
    server.close()
    event_loop.run_until_complete(server.wait_closed())
    event_loop.close()

ran_for = time.time() - stats['started at']
print(f"伺服器執行時間：{ran_for} 秒")

內容解密：

上述程式碼定義了非同步伺服器的連線建立和資料接收邏輯。當連線建立時，更新傳輸物件和連線資料。當接收到資料時，將其轉換為大寫並發送回給客戶端。伺服器啟動後，會持續執行直到接收到鍵盤中斷訊號。最終，關閉伺服器和事件迴圈，並計算伺服器執行時間。

圖表翻譯：

@startuml
skinparam backgroundColor #FEFEFE
skinparam componentStyle rectangle

title Asyncio 非同步網路伺服器與客戶端實作

package "統計分析流程" {
    package "資料收集" {
        component [樣本資料] as sample
        component [母體資料] as population
    }

    package "描述統計" {
        component [平均數/中位數] as central
        component [標準差/變異數] as dispersion
        component [分佈形狀] as shape
    }

    package "推論統計" {
        component [假設檢定] as hypothesis
        component [信賴區間] as confidence
        component [迴歸分析] as regression
    }
}

sample --> central : 計算
sample --> dispersion : 計算
central --> hypothesis : 檢驗
dispersion --> confidence : 估計
hypothesis --> regression : 建模

note right of hypothesis
  H0: 虛無假設
  H1: 對立假設
  α: 顯著水準
end note

@enduml

上述圖表描述了連線建立和資料接收的流程。當連線建立時，更新傳輸物件和連線資料。當接收到資料時，轉換資料為大寫並發送回給客戶端，同時更新訊息資料。

瞭解Asyncio的優勢和實際應用

Asyncio是一種強大的Python函式庫，能夠幫助開發者建立高效能的非同步網路客戶端和伺服器。透過使用Asyncio，開發者可以輕鬆地處理複雜的網路通訊任務，並且可以混合使用不同的非同步工作負載，使得框架更加強大。

Asyncio的優勢

Asyncio具有許多優勢，包括：

  • 高效能：Asyncio可以處理大量的網路連線和訊息，且效能優異。
  • 簡單易用：Asyncio的API簡單易用，開發者可以快速上手。
  • 混合工作負載：Asyncio可以混合使用不同的非同步工作負載，使得框架更加強大。

從效能最佳化視角來看，asyncio 為 Python 網路程式設計提供了高效的非同步解決方案。本文深入剖析了利用 asyncio 建構 TCP 和 Web 伺服器的實務技巧，並展示瞭如何收集伺服器效能統計資料。asyncio 的核心優勢在於其單執行緒非同步模型，能有效處理大量 I/O 密集型任務，大幅提升伺服器吞吐量。然而，程式碼的非同步特性也增加了除錯的複雜度，需要開發者熟悉非同步程式設計的思維模式。

相較於傳統的多執行緒或多程式模型，asyncio 避免了執行緒或程式切換的開銷，從而降低了系統資源消耗，並提升了程式碼執行效率。尤其在高併發場景下，asyncio 的優勢更加明顯。此外，aiohttp 等根據 asyncio 的非同步 Web 框架簡化了 Web 伺服器的開發流程，讓開發者能更專注於業務邏輯的實作。不過，需要注意的是，asyncio 並非所有場景的最佳選擇，對於 CPU 密集型任務，多程式模型可能更為合適。

展望未來，隨著 Python 生態系統的持續發展，預計 asyncio 將扮演更重要的角色。更多根據 asyncio 的函式庫和框架將不斷湧現，進一步降低非同步程式設計的門檻。同時，Python 語言本身也將持續最佳化 asyncio 的效能和易用性，使其更廣泛地應用於網路程式設計、資料科學等領域。對於追求高效能網路應用的開發者而言，asyncio 值得深入學習和應用。玄貓認為，掌握 asyncio 將成為 Python 開發者的核心競爭力之一，尤其在構建高效能、高併發的網路服務方面更具優勢。