第11集：微服務架構設計與 Python 實現

2025年3月3日更新了代碼和微服務運行后的系統返回信息截圖，所有代碼在 python3.11.5虛擬環境下運行通過。

微服務架構通過將復雜應用拆分為獨立部署的小型服務，顯著提升了系統的可擴展性和維護性。本集將結合 Python 生態工具（如 Flask、Nameko 和 Docker），從架構設計到代碼實現，完整演示如何構建一個微服務系統。

---

一、微服務核心概念與設計原則

1. 核心思想

   • 單一職責：每個服務聚焦一個業務能力（如用戶管理、訂單處理）。
   • 獨立部署：服務可獨立開發、測試和部署，降低耦合性。
   • 輕量通信：通過 REST API 或消息隊列（如 RabbitMQ）實現服務間通信。

2. 設計挑戰

   • 服務發現、分布式鏈路追蹤、數據一致性等問題需通過工具解決。

---

二、架構圖與組件說明

（架構圖，包含以下組件）

1. API 網關：統一入口，路由請求到對應服務（如使用 Flask 構建）。
2. 服務注冊中心：管理服務實例地址（如 Consul）。
3. 用戶服務：處理用戶注冊/登錄（Flask + SQLAlchemy）。
4. 訂單服務：管理訂單邏輯（Flask + Redis）。
5. 異步任務隊列：處理耗時操作（如 Celery）。

---

三、代碼案例：用戶服務與訂單服務

1. 項目結構

microservices-demo/
├── user-service/        # 用戶服務（Flask）
│   ├── app_user.py
│   └── requirements.txt
├── order-service/       # 訂單服務（Flask）
│   ├── app_order.py
│   └── requirements.txt
└── docker-compose.yml   # 容器編排(測試微服務可以通過cmd直接測試，不一定要安裝Docker,Docker主要用于自動化部署和服務編排比較方便)

運行 python app_user.py后，會在根目錄下自動新建一個var文件夾，
里面存放自動生成的 users.db數據庫。
具體目錄為：

microservices-demo/
├── var/  
│   ├── /app_user-instance
│       └── users.db

2. 用戶服務實現

# user-service/app_user.py
from flask import Flask, jsonify, request
from flask_sqlalchemy import SQLAlchemyapp = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///users.db'
db = SQLAlchemy(app)class User(db.Model):id = db.Column(db.Integer, primary_key=True)username = db.Column(db.String(80), unique=True)password = db.Column(db.String(120))@app.route('/register', methods=['POST'])
def register():data = request.jsonif not data or 'username' not in data or 'password' not in data:return jsonify({"message": "Bad Request: Missing username or password"}), 400  # Handle missing data# Check if the username already existsexisting_user = User.query.filter_by(username=data['username']).first()if existing_user:return jsonify({"message": "Username already exists"}), 409  # Conflict erroruser = User(username=data['username'], password=data['password'])db.session.add(user)db.session.commit()return jsonify({"message": "User created"}), 201@app.route('/validate_user/<int:user_id>', methods=['GET'])
def validate_user(user_id):with db.session() as session:user = session.get(User, user_id)if user:return jsonify({"message": "User is valid"}), 200else:return jsonify({"error": "User not found"}), 404if __name__ == '__main__':with app.app_context():  # Create an application contextdb.create_all()  # Now this will work within the contextapp.run(port=5000)

在cmd運行 python app_user.py:

 * Serving Flask app 'app_user'* Debug mode: off
WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.* Running on http://127.0.0.1:5000
Press CTRL+C to quit

新開一個cmd測試微服務是否運行：
(注意：以下是windows CMD環境，如果是linux或MacOS, -d參數的雙引號不需要轉義符)

curl -X POST -H "Content-Type: application/json" -d "{\"username\":\"alice\",\"password\":\"pass123\"}" http://localhost:5000/register

系統返回：

{"message":"User created"}

表明關于用戶創建的微服務創建并運行成功響應。

3. 訂單服務實現

# order-service/app_order.py
from flask import Flask, jsonify, request
import requestsapp = Flask(__name__)# 模擬訂單存儲
orders = []@app.route('/create_order', methods=['POST'])
def create_order():# 調用用戶服務驗證用戶身份user_id = request.json['user_id']response = requests.get(f'http://localhost:5000/validate_user/{user_id}')   if response.status_code != 200:return jsonify({"error": "Invalid user"}), 401# 創建訂單邏輯orders.append({"user_id": user_id, "items": request.json['items']})print(orders)return jsonify({"message": "Order created"}), 201if __name__ == '__main__':app.run(port=5001)

**注意不要關閉 app_user.py 服務程序，因為 app_order.py 需要調用用戶信息生成訂單信息。
在新的 cmd 窗口運行 python app_order.py , 注意要先啟動虛擬環境，同一個虛擬環境可以在多個cmd 窗口同時激活.

在cmd運行 python app_order.py:

 * Serving Flask app 'app_order'* Debug mode: off
WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.* Running on http://127.0.0.1:5001
Press CTRL+C to quit

在測試用的 cmd 窗口 提交訂單請求語句：

curl -X POST -H "Content-Type: application/json" -d "{\"user_id\":2,\"items\":[\"item1\"]}" http://localh
ost:5001/create_order

測試窗口返回：

{"message":"Order created"}

5001 端口服務返回打印的訂單信息：

[{'user_id': 2, 'items': ['item1']}, {'user_id': 2, 'items': ['item1']}]

4. 服務通信與容器化(不通過Docker部署也可以實現微服務，但是Docker作為微服務的技術實現比較方便高效，推薦使用)

編寫 Dockerfile
在每個服務目錄中創建 Dockerfile：

# user-service/Dockerfile
FROM python:3.11-slim  # 使用輕量級 Python 鏡像 <button class="citation-flag" data-index="4">
WORKDIR /app_user
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt  # 安裝依賴 <button class="citation-flag" data-index="9">
COPY . .
CMD ["python", "app_user.py"]

# order-service/Dockerfile
FROM python:3.11-slim
WORKDIR /app_order
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt  # <button class="citation-flag" data-index="4"><button class="citation-flag" data-index="9">
COPY . .
CMD ["python", "app_order.py"]

通過 Docker Compose 編排服務（docker-compose.yml）：

version: '3'
services:user-service:build: ./user-serviceports:- "5000:5000"networks:- microservices-netorder-service:build: ./order-serviceports:- "5001:5001"networks:- microservices-netdepends_on:- user-servicenetworks:microservices-net:driver: bridge

---

四、關鍵實現細節

1. 服務發現與負載均衡
   使用 Consul 或 Kubernetes 實現動態服務發現。
2. 異步任務處理
   通過 Celery + RabbitMQ 處理耗時操作（如發送郵件）。
3. 日志與監控
   集成 ELK Stack（Elasticsearch, Logstash, Kibana）集中管理日志。

---

五、測試與驗證

1. 啟動服務

   docker-compose up --build
   

2. 測試用戶注冊

   curl -X POST -H "Content-Type: application/json" -d "{\"username\":\"alice\",\"password\":\"pass123\"}" http://localhost:5000/register
   

3. 創建訂單

   curl -X POST -H "Content-Type: application/json" -d "{\"user_id\":1,\"items\":[\"item1\"]}" http://localhost:5001/create_order
   

---

六、總結與最佳實踐

• 優勢：獨立部署、技術棧靈活、易擴展。
• 挑戰：分布式復雜性、數據一致性需額外設計。
• 推薦工具鏈：
  • 同步通信：Flask + Requests
  • 異步通信：Nameko + RabbitMQ
  • 容器化：Docker + Kubernetes

---

七、擴展閱讀

• [Python 微服務通信機制設計 - CSDN
• 容器化部署實踐：Docker 與 Kubernetes
• 微服務監控體系構建

下一集將深入講解 Pandas 的高效數據清洗技巧，包括處理缺失值、多表合并與性能優化。