《Python实战进阶》No 11：微服务架构设计与 Python 实现

第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"}), [e-mail]('/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 的高效数据清洗技巧，包括处理缺失值、多表合并与性能优化。