FastAPI 非同步程式設計與微服務架構中的會話管理

– 8 分鐘閱讀

FastAPI 作為一個高效能的 Python Web 框架，其非同步特性在構建高併發應用時至關重要。本文從 Coroutine 的基本應用開始，逐步深入探討非同步交易、HTTP/2 協定的整合，以及如何利用 Celery 實作非同步任務排程和分散式任務佇列。此外，針對微服務架構中常見的會話管理挑戰，本文分析了根據 Cookie 和 Token 的會話管理機制，並提供瞭如何在 Flask 應用中安全地建立、管理和刪除使用者會話的實務程式碼範例。最後，本文也涵蓋瞭如何使用 RabbitMQ 和 Kafka 建立訊息驅動架構，以及如何運用 SSE 和 WebSocket 技術實作伺服器推播事件和全雙工通訊，以提升應用程式的即時性和互動性。

技術要求

在開始之前，請確保您已經安裝了FastAPI和相關的依賴項。您可以使用pip安裝FastAPI：

pip install fastapi

實作Coroutine

Coroutine是一種特殊的函式，可以暫停和還原執行，允許其他Coroutine執行。FastAPI提供了對Coroutine的內建支援，允許您輕鬆地建立和管理Coroutine。

from fastapi import FastAPI

app = FastAPI()

async def my_coroutine():
    # 執行一些非同步操作
    await asyncio.sleep(1)
    return {"message": "Coroutine完成"}

@app.get("/coroutine")
async def read_coroutine():
    return await my_coroutine()

應用Coroutine切換

Coroutine切換是指在多個Coroutine之間切換執行的過程。FastAPI提供了asyncio函式庫來支援Coroutine切換。

import asyncio

async def my_coroutine1():
    # 執行一些非同步操作
    await asyncio.sleep(1)
    return {"message": "Coroutine 1完成"}

async def my_coroutine2():
    # 執行一些非同步操作
    await asyncio.sleep(2)
    return {"message": "Coroutine 2完成"}

async def main():
    task1 = asyncio.create_task(my_coroutine1())
    task2 = asyncio.create_task(my_coroutine2())
    results = await asyncio.gather(task1, task2)
    return results

@app.get("/coroutines")
async def read_coroutines():
    return await main()

設計非同步交易

非同步交易是指在多個Coroutine之間執行的交易。FastAPI提供了asyncio函式庫來支援非同步交易。

import asyncio

async def my_transaction():
    # 執行一些非同步操作
    await asyncio.sleep(1)
    return {"message": "交易完成"}

async def main():
    task = asyncio.create_task(my_transaction())
    result = await task
    return result

@app.get("/transaction")
async def read_transaction():
    return await main()

使用HTTP/2協定

HTTP/2是一種新的HTTP協定，提供了更好的效能和效率。FastAPI提供了對HTTP/2的內建支援，允許您輕鬆地使用HTTP/2協定。

from fastapi import FastAPI

app = FastAPI()

@app.get("/http2")
async def read_http2():
    return {"message": "HTTP/2完成"}

建立非同步背景任務

非同步背景任務是指在背景執行的非同步操作。FastAPI提供了asyncio函式庫來支援非同步背景任務。

import asyncio

async def my_background_task():
    # 執行一些非同步操作
    await asyncio.sleep(1)
    return {"message": "背景任務完成"}

async def main():
    task = asyncio.create_task(my_background_task())
    result = await task
    return result

@app.get("/background-task")
async def read_background_task():
    return await main()

使用Coroutine

Coroutine是一種特殊的函式，可以暫停和還原執行，允許其他Coroutine執行。FastAPI提供了對Coroutine的內建支援，允許您輕鬆地建立和管理Coroutine。

from fastapi import FastAPI

app = FastAPI()

async def my_coroutine():
    # 執行一些非同步操作
    await asyncio.sleep(1)
    return {"message": "Coroutine完成"}

@app.get("/coroutine")
async def read_coroutine():
    return await my_coroutine()

建立多個任務

多個任務是指在多個Coroutine之間執行的多個任務。FastAPI提供了asyncio函式庫來支援多個任務。

import asyncio

async def my_task1():
    # 執行一些非同步操作
    await asyncio.sleep(1)
    return {"message": "任務 1完成"}

async def my_task2():
    # 執行一些非同步操作
    await asyncio.sleep(2)
    return {"message": "任務 2完成"}

async def main():
    task1 = asyncio.create_task(my_task1())
    task2 = asyncio.create_task(my_task2())
    results = await asyncio.gather(task1, task2)
    return results

@app.get("/tasks")
async def read_tasks():
    return await main()

Understanding Celery任務

Celery是一種分散式任務佇列，允許您輕鬆地建立和管理任務。FastAPI提供了對Celery的內建支援，允許您輕鬆地使用Celery任務。

from celery import Celery

app = Celery("my_app")

@app.task
def my_task():
    # 執行一些非同步操作
    return {"message": "任務完成"}

@app.get("/celery-task")
async def read_celery_task():
    return await my_task.delay()

設定任務排程系統

要建立一個任務排程系統，首先需要建立和組態 Celery 例項。這個過程包括初始化 Celery、定義任務以及組態 worker 伺服器。

建立 Celery 例項

建立 Celery 例項是任務排程系統的基礎。這一步驟包括初始化 Celery、設定 broker 和結果後端。以下是建立 Celery 例項的基本步驟：

from celery import Celery

app = Celery('tasks', broker='amqp://guest@localhost//')

在這個例子中，建立了一個名為 tasks 的 Celery 例項，使用 RabbitMQ 作為 broker。

定義任務

定義任務是任務排程系統的核心。任務是可以被 Celery 執行的函式。以下是定義一個任務的例子：

@app.task
def add(x, y):
    return x + y

在這個例子中，定義了一個名為 add 的任務，該任務接收兩個引數 x 和 y，並傳回它們的和。

執行任務

執行任務是任務排程系統的最終目標。以下是執行任務的例子：

result = add.delay(4, 4)
print(result.get())  # outputs: 8

在這個例子中，執行了 add 任務，傳入引數 4 和 4，並列印預出結果。

啟動 worker 伺服器

啟動 worker 伺服器是任務排程系統的必要步驟。以下是啟動 worker 伺服器的例子：

celery -A tasks worker --loglevel=info

在這個例子中，啟動了 worker 伺服器，使用 tasks 模組，log 級別設為 info。

監控任務

監控任務是任務排程系統的重要部分。以下是監控任務的例子：

from celery import result

result = add.delay(4, 4)
print(result.status)  # outputs: success

在這個例子中，監控了 add 任務的狀態，列印預出結果。

使用 RabbitMQ 建立訊息驅動交易

RabbitMQ 是一個訊息代理伺服器，可以用於建立訊息驅動交易。以下是使用 RabbitMQ 建立訊息驅動交易的步驟：

from celery import Celery
from kombu import Connection, Exchange, Queue

app = Celery('tasks', broker='amqp://guest@localhost//')

exchange = Exchange('my_exchange', type='direct')
queue = Queue('my_queue', exchange, routing_key='my_key')

connection = Connection('amqp://guest@localhost//')
channel = connection.channel()

channel.queue_declare(queue=queue, durable=True)
channel.exchange_declare(exchange=exchange, durable=True)
channel.queue_bind(queue=queue, exchange=exchange, routing_key='my_key')

在這個例子中，建立了一個名為 my_exchange 的 exchange 和一個名為 my_queue 的 queue，使用 RabbitMQ 作為 broker。

使用 Kafka 建立釋出/訂閱訊息

Kafka 是一個釋出/訂閱訊息系統，可以用於建立釋出/訂閱訊息。以下是使用 Kafka 建立釋出/訂閱訊息的步驟：

from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers='localhost:9092')

topic = 'my_topic'
message = 'Hello, World!'

producer.send(topic, value=message.encode('utf-8'))

在這個例子中，建立了一個名為 my_topic 的 topic，使用 Kafka 作為 broker，釋出了一條訊息。

執行 Kafka 代理和伺服器

執行 Kafka 代理和伺服器是使用 Kafka 的必要步驟。以下是執行 Kafka 代理和伺服器的例子：

kafka-server-start.sh

在這個例子中，啟動了 Kafka 代理和伺服器。

建立主題

建立主題是使用 Kafka 的必要步驟。以下是建立主題的例子：

kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 my_topic

在這個例子中，建立了一個名為 my_topic 的主題，使用 Kafka 作為 broker。

實作釋出者

實作釋出者是使用 Kafka 的必要步驟。以下是實作釋出者的例子：

from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers='localhost:9092')

topic = 'my_topic'
message = 'Hello, World!'

producer.send(topic, value=message.encode('utf-8'))

在這個例子中，實作了一個釋出者，使用 Kafka 作為 broker，釋出了一條訊息。

執行消費者

執行消費者是使用 Kafka 的必要步驟。以下是執行消費者的例子：

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my_topic

在這個例子中，啟動了一個消費者，使用 Kafka 作為 broker，消費了一條訊息。

非同步程式設計實作

非同步程式設計是一種可以提高程式效率和回應速度的技術。以下是幾種非同步程式設計的實作方法：

1. 伺服器推播事件（Server-Sent Events, SSE）

SSE是一種允許伺服器主動推播事件給客戶端的技術。客戶端可以透過SSE接收伺服器推播的事件，並進行相應的處理。

import asyncio
import websockets

async def handle_connection(websocket, path):
    # 伺服器推播事件
    await websocket.send("Hello, client!")

async def main():
    async with websockets.serve(handle_connection, "localhost", 8765):
        await asyncio.Future()  # run forever

asyncio.run(main())

2. 非同步WebSocket

WebSocket是一種允許客戶端和伺服器之間進行全雙工通訊的技術。非同步WebSocket可以提高通訊的效率和回應速度。

import asyncio
import websockets

async def handle_connection(websocket, path):
    # 客戶端和伺服器之間的通訊
    while True:
        message = await websocket.recv()
        await websocket.send(f"Received: {message}")

async def main():
    async with websockets.serve(handle_connection, "localhost", 8765):
        await asyncio.Future()  # run forever

asyncio.run(main())

3. 反應式程式設計

反應式程式設計是一種根據事件驅動的程式設計風格。它可以提高程式的效率和回應速度。

import asyncio

async def task1():
    # 任務1
    await asyncio.sleep(1)
    print("Task 1 finished")

async def task2():
    # 任務2
    await asyncio.sleep(2)
    print("Task 2 finished")

async def main():
    # 執行任務
    await asyncio.gather(task1(), task2())

asyncio.run(main())

4. 背景程式

背景程式可以提高程式的效率和回應速度。

import asyncio

async def background_task():
    # 背景程式
    await asyncio.sleep(1)
    print("Background task finished")

async def main():
    # 執行背景程式
    asyncio.create_task(background_task())
    await asyncio.sleep(2)

asyncio.run(main())

5. API資源存取

API資源存取可以提高程式的效率和回應速度。

import asyncio
import aiohttp

async def fetch_api():
    # API資源存取
    async with aiohttp.ClientSession() as session:
        async with session.get("https://api.example.com/data") as response:
            data = await response.json()
            print(data)

async def main():
    # 執行API資源存取
    await fetch_api()

asyncio.run(main())

6. 事件自訂

事件自訂可以提高程式的效率和回應速度。

import asyncio

async def custom_event():
    # 事件自訂
    await asyncio.sleep(1)
    print("Custom event finished")

async def main():
    # 執行事件自訂
    asyncio.create_task(custom_event())
    await asyncio.sleep(2)

asyncio.run(main())

7. 啟動事件

啟動事件可以提高程式的效率和回應速度。

import asyncio

async def startup_event():
    # 啟動事件
    await asyncio.sleep(1)
    print("Startup event finished")

async def main():
    # 執行啟動事件
    asyncio.create_task(startup_event())
    await asyncio.sleep(2)

asyncio.run(main())

會話管理與微服務架構

在開發複雜的網路應用時，會話管理是一個重要的議題。會話管理涉及到如何建立、管理和刪除使用者的會話，以確保使用者的資料安全和應用的穩定性。以下將介紹如何在微服務架構中實作會話管理。

建立使用者會話

建立使用者會話的過程通常涉及到使用者登入和驗證。當使用者成功登入後，應用需要建立一個會話來儲存使用者的資料。這可以透過使用根據 cookie 的會話管理或 token-based 的會話管理來實作。

from flask import Flask, session
from flask_session import Session

app = Flask(__name__)
app.config["SESSION_PERMANENT"] = False
app.config["SESSION_TYPE"] = "filesystem"
Session(app)

@app.route("/login", methods=["POST"])
def login():
    # 驗證使用者登入資訊
    username = request.form["username"]
    password = request.form["password"]
    if validate_login(username, password):
        # 建立使用者會話
        session["username"] = username
        return "登入成功"
    else:
        return "登入失敗"

管理會話資料

一旦會話被建立，應用需要管理會話資料以確保使用者的資料安全。這可以透過使用加密和儲存會話資料在安全的位置來實作。

from flask import session

@app.route("/user_data", methods=["GET"])
def get_user_data():
    # 從會話中取得使用者資料
    username = session.get("username")
    if username:
        # 從資料函式庫中取得使用者資料
        user_data = get_user_data_from_db(username)
        return user_data
    else:
        return "使用者未登入"

刪除會話

當使用者登出時，應用需要刪除會話以確保使用者的資料安全。這可以透過使用根據 cookie 的會話管理或 token-based 的會話管理來實作。

from flask import session

@app.route("/logout", methods=["POST"])
def logout():
    # 刪除會話
    session.pop("username", None)
    return "登出成功"

從系統架構的視角來看，FastAPI 提供了強大的非同步程式設計能力，能有效提升網路應用程式的效能和反應速度。本文涵蓋了從 Coroutine 的基本應用到結合 Celery、RabbitMQ 和 Kafka 等訊息佇列系統的整合方案，展現了 FastAPI 在處理非同步任務、交易和高併發情境下的優勢。然而，FastAPI 的非同步特性也對開發者提出了更高的要求，需要深入理解事件迴圈、任務管理和非同步程式設計的最佳實務。此外，在微服務架構中，會話管理的複雜度也隨之提升，需要仔細考量不同方案的優劣，例如根據 Cookie 或 Token 的機制，並搭配適當的加密和儲存策略以確保安全性。展望未來，隨著 Python 非同步生態的持續發展，FastAPI 的應用場景將更加廣泛，預期將在高效能網路服務、即時資料串流處理和分散式系統等領域扮演更重要的角色。對於追求高效能和可擴充套件性的開發團隊而言，深入學習和掌握 FastAPI 的非同步特性將是提升競爭力的關鍵。

玄貓

玄貓

技術愛好者，專注於分享程式開發、雲端技術與 AI 應用的心得體會。