微服務架構日漸普及，開發者將應用程式拆解成小型獨立服務，各自獨立開發、佈署和擴充套件。本文除了介紹 gRPC 和 Lanarky 兩種微服務實作方式外，也探討事件溯源模式，以利讀者更全面地理解現代軟體架構設計。gRPC 範例中，我們以 Python 程式碼示範如何定義 protobuf 介面以及伺服器端和客戶端的實作。Lanarky 範例則展示瞭如何利用 FastAPI 和 OpenAI 建立 LLM 微服務。此外，我們也探討了無伺服器模式，並以 AWS Lambda 函式示範其實作流程。最後，我們深入講解事件溯源模式，包含其核心元件、手動實作和使用 eventsourcing 函式庫的實作，並分析其在系統狀態管理方面的優勢。
微服務架構設計模式
在現代軟體開發中，微服務架構已成為一種流行的設計模式。它允許開發者將應用程式分解為多個小型、獨立的服務，每個服務都可以獨立開發、佈署和擴充套件。在本章中，我們將探討幾種常見的微服務架構設計模式，包括 gRPC 和 Lanarky。
使用 gRPC 實作微服務模式
gRPC 是一種高效的遠端程式呼叫（RPC）框架，適用於微服務之間的通訊。它使用 Protocol Buffers（protobuf）作為介面定義語言（IDL），並支援多種程式語言。
gRPC 微服務範例
以下是一個簡單的 gRPC 微服務範例，使用 Python 語言實作。首先，我們需要定義服務介面和訊息格式：
syntax = "proto3";
package payment;
service PaymentService {
rpc ProcessPayment(PaymentRequest) returns (PaymentResponse) {}
}
message PaymentRequest {
float amount = 1;
string currency = 2;
string user_id = 3;
}
message PaymentResponse {
string status = 1;
}
接下來，我們實作伺服器端程式碼：
from concurrent import futures
import grpc
import payment_pb2
import payment_pb2_grpc
class PaymentService(payment_pb2_grpc.PaymentServiceServicer):
def ProcessPayment(self, request, context):
# 處理支付請求
return payment_pb2.PaymentResponse(status="SUCCESS")
def serve():
server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
payment_pb2_grpc.add_PaymentServiceServicer_to_server(PaymentService(), server)
server.add_insecure_port('[::]:50051')
server.start()
print("Payment Processing Service ready!")
server.wait_for_termination()
if __name__ == '__main__':
serve()
客戶端程式碼如下：
import grpc
import payment_pb2
import payment_pb2_grpc
def run():
channel = grpc.insecure_channel('localhost:50051')
stub = payment_pb2_grpc.PaymentServiceStub(channel)
request = payment_pb2.PaymentRequest(amount=99.99, currency="USD", user_id="user456")
response = stub.ProcessPayment(request)
print("Payment Service responded.")
print(f"Response status: {response.status}")
if __name__ == '__main__':
run()
程式碼解密：
- 在伺服器端程式碼中，我們定義了一個
PaymentService類別，實作了ProcessPayment方法來處理支付請求。 - 在客戶端程式碼中，我們建立了一個 gRPC 通道，並使用
PaymentServiceStub類別來呼叫ProcessPayment方法。
使用 Lanarky 實作 LLM 服務
Lanarky 是一個根據 FastAPI 的網頁框架，用於建置使用大語言模型（LLM）的微服務。
LLM 服務範例
以下是一個簡單的 LLM 服務範例，使用 Lanarky 和 OpenAI 的 API。首先，我們需要安裝 Lanarky 和相關依賴套件：
pip install lanarky uvicorn
接下來，我們實作 LLM 服務程式碼：
import os
import uvicorn
from lanarky import Lanarky
from lanarky.adapters.openai.resources import ChatCompletionResource
from lanarky.adapters.openai.routing import OpenAIAPIRouter
app = Lanarky()
router = OpenAIAPIRouter()
@router.post("/chat")
def chat(stream: bool = True) -> ChatCompletionResource:
system = "Here is your assistant"
return ChatCompletionResource(stream=stream, system=system)
if __name__ == "__main__":
app.include_router(router)
uvicorn.run(app)
客戶端程式碼如下：
import click
import sys
from lanarky.clients import StreamingClient
args = sys.argv[1:]
if len(args) == 1:
message = args[0]
client = StreamingClient()
for event in client.stream_response(
"POST",
"/chat",
params={"stream": "false"},
json={"messages": [dict(role="user", content=message)]},
):
print(f"{event.event}: {event.data}")
else:
print("You need to pass a message!")
程式碼解密：
- 在伺服器端程式碼中，我們定義了一個
/chat路由，使用ChatCompletionResource類別來處理聊天請求。 - 在客戶端程式碼中，我們使用
StreamingClient類別來呼叫/chat路由，並傳遞使用者訊息。
無伺服器模式
無伺服器模式是一種雲端運算模式，允許開發者專注於撰寫程式碼，而無需管理伺服器。雲端提供商會根據事件觸發器自動擴充套件和執行程式碼。
無伺服器模式的應用場景
- 自動備份資料：無伺服器函式可以排程自動備份重要資料到雲端儲存。
- 影像處理：當使用者上傳影像時，無伺服器函式可以自動調整影像大小、壓縮或套用濾鏡。
- 電子商務收據生成：當購買完成後，無伺服器函式可以生成 PDF 收據並透過電子郵件傳送給客戶。
graph LR;
A[使用者請求] --> B[無伺服器函式];
B --> C[事件觸發器];
C --> D[雲端提供商];
D --> E[自動擴充套件];
E --> F[執行程式碼];
圖表翻譯：
此圖示呈現了無伺服器模式的流程。首先，使用者發出請求，觸發無伺服器函式。無伺服器函式會根據事件觸發器自動執行。雲端提供商負責管理無伺服器函式的執行環境，並根據需求自動擴充套件資源。最終，無伺服器函式會執行相關的程式碼，完成指定的任務。
程式碼解密：
- 無伺服器模式允許開發者專注於撰寫程式碼，而無需管理伺服器。
- 事件觸發器是無伺服器模式的核心，可以根據不同的事件自動觸發無伺服器函式的執行。
無伺服器架構模式（Serverless Pattern）與事件溯源模式（Event Sourcing Pattern）
無伺服器架構模式簡介
無伺服器架構模式是一種雲端運算的執行模型，它允許開發者編寫和佈署應用程式而無需管理底層的基礎設施。這種模式特別適用於事件驅動的架構和微服務架構。
無伺服器架構模式的應用場景
- 事件驅動架構：無伺服器架構非常適合處理事件驅動的應用程式，例如圖片處理、動態PDF生成等。
- 微服務架構：每個微服務都可以被實作為一個無伺服器函式，從而簡化管理和擴充套件。
實作無伺服器架構模式
讓我們使用AWS Lambda來建立一個簡單的無伺服器函式，該函式計算一個數字的平方。
步驟1：編寫Python程式碼
首先，我們需要編寫Python程式碼來實作Lambda函式。我們建立了一個名為lambda_handler的函式，它接受兩個引數：event和context。在我們的例子中，輸入數字是透過event字典中的"number"鍵來存取的。我們計算該數字的平方並傳回結果。
import json
def lambda_handler(event, context):
number = event["number"]
squared = number * number
return f"The square of {number} is {squared}."
#### 內容解密：
此段程式碼定義了一個AWS Lambda函式，名為lambda_handler。該函式接受兩個引數：event和context。其中，event是一個字典，包含了觸發Lambda函式的事件資訊。在這個例子中，我們假設event字典中有一個名為"number"的鍵，其值為需要計算平方的數字。函式首先從event字典中提取這個數字，然後計算其平方，最後傳回一個字串，描述了輸入數字及其平方的結果。
步驟2：佈署Lambda函式
為了佈署Lambda函式，我們首先需要將Python程式碼檔案壓縮成ZIP檔案。然後，使用awslocal工具將Lambda函式佈署到本地的AWS基礎設施（透過LocalStack模擬）。
zip lambda.zip lambda_function_square.py
awslocal lambda create-function \
--function-name lambda_function_square \
--runtime python3.11 \
--zip-file fileb://lambda.zip \
--handler lambda_function_square.lambda_handler \
--role arn:aws:iam::000000000000:role/lambda-role
#### 內容解密：
這段指令首先將Python指令碼檔案lambda_function_square.py壓縮成名為lambda.zip的ZIP檔案。接著，使用awslocal命令列工具在本地模擬的AWS環境中建立一個Lambda函式。該Lambda函式被命名為lambda_function_square，執行環境為Python 3.11，程式碼來自剛才建立的ZIP檔案，處理函式為lambda_function_square.lambda_handler，並指定了一個虛擬的IAM角色。
步驟3：測試Lambda函式
我們可以使用awslocal工具測試Lambda函式，提供一個包含輸入數字（例如6）的JSON檔案作為輸入。
awslocal lambda invoke --function-name lambda_function_square \
--payload file://payload.json output.txt
#### 內容解密：
此命令用於呼叫前面建立的Lambda函式，並傳遞一個包含輸入資料的JSON檔案（位於payload.json）作為事件資料。執行結果將被儲存到名為output.txt的檔案中。
事件溯源模式簡介
事件溯源模式透過將狀態變更儲存為一系列事件來實作狀態的重建和稽核追蹤。這種模式在系統狀態複雜且業務規則複雜的情況下尤其有用。
事件溯源模式的應用場景
- 財務交易：記錄每筆金融交易的變更，以提供透明、可稽核和安全的財務活動記錄。
- 庫存管理：跟蹤每個物品的生命週期，以維護準確的庫存記錄，並預測未來的庫存需求。
事件溯源模式：深入解析與實作
事件溯源（Event Sourcing）是一種軟體架構模式，主要用於捕捉和儲存系統中發生的所有變更，以事件的形式記錄下來。這種模式在需要追蹤和分析系統狀態變更的應用中非常有用，例如客戶行為分析、庫存管理等。
事件溯源模式的元件
事件溯源模式的實作包含以下幾個關鍵元件：
- 事件（Event）：代表系統狀態的變更，包含事件型別和相關資料。事件一旦被建立和套用，就不能被更改。
- 聚合（Aggregate）：代表一個業務邏輯或資料的單元，負責追蹤和管理相關事件。
- 事件儲存（Event Store）：儲存所有發生的事件，提供事件的歷史記錄。
手動實作事件溯源模式
首先，我們來看看如何手動實作事件溯源模式。以下是一個銀行帳戶的例子：
class Account:
def __init__(self):
self.balance = 0
self.events = []
def apply_event(self, event):
if event["type"] == "deposited":
self.balance += event["amount"]
elif event["type"] == "withdrawn":
self.balance -= event["amount"]
self.events.append(event)
def deposit(self, amount):
event = {"type": "deposited", "amount": amount}
self.apply_event(event)
def withdraw(self, amount):
event = {"type": "withdrawn", "amount": amount}
self.apply_event(event)
內容解密：
此程式碼定義了一個 Account 類別，代表一個銀行帳戶。它包含了 balance 和 events 兩個屬性，分別用於記錄帳戶餘額和所有發生的事件。apply_event 方法根據事件型別更新帳戶餘額並儲存事件。deposit 和 withdraw 方法分別用於存款和提款，並呼叫 apply_event 方法來更新帳戶狀態。
使用函式庫實作事件溯源模式
接下來，我們使用 eventsourcing 函式庫來實作事件溯源模式。以下是一個庫存管理的例子：
from eventsourcing.domain import Aggregate, event
from eventsourcing.application import Application
class InventoryItem(Aggregate):
@event("ItemCreated")
def __init__(self, name, quantity=0):
self.name = name
self.quantity = quantity
@event("QuantityIncreased")
def increase_quantity(self, amount):
self.quantity += amount
@event("QuantityDecreased")
def decrease_quantity(self, amount):
self.quantity -= amount
內容解密：
此程式碼定義了一個 InventoryItem 類別，代表一個庫存專案。它繼承自 Aggregate 類別，並使用了 @event 裝飾器來定義事件。__init__ 方法初始化庫存專案，increase_quantity 和 decrease_quantity 方法分別用於增加和減少庫存數量。
事件溯源模式的優點
事件溯源模式提供了許多優點，包括：
- 可追蹤性：事件溯源模式提供了一個完整的事件歷史記錄，使得系統狀態的變更可以被追蹤和分析。
- 可擴充套件性：事件溯源模式使得系統可以更容易地擴充套件和修改，因為新的事件型別可以被輕易地新增。
- 靈活性：事件溯源模式提供了很大的靈活性，因為系統可以根據不同的事件型別進行不同的處理。
