Python 微服务架构：Flask + Docker + Redis + RabbitMQ 实战

小飞兽 Python 8 次阅读 2026-07-24

Introduction

微服务是现代后端架构的主流模式。本文从拆分 Flask 单体应用讲起，讲解服务间通信（HTTP/REST、消息队列）、Docker 容器化部署、Redis 缓存、RabbitMQ 消息队列的集成。

---

单体到微服务拆分

假设原有单体应用包含用户、订单、库存三个模块。拆分为三个独立服务。

服务间通信：HTTP REST

# order_service.py
import requests
from flask import Flask, jsonify

app = Flask(__name__)

@app.route('/orders/<int:order_id>')
def get_order(order_id):
    # 调用用户服务获取用户信息
    user = requests.get(f'http://user-service:5001/users/{order_id}').json()
    return jsonify({'order_id': order_id, 'user': user, 'status': 'shipped'})

Redis 缓存

pip install redis

import redis, json

r = redis.Redis(host='redis', port=6379, decode_responses=True)

def get_user(user_id):
    cache_key = f'user:{user_id}'
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)

    user = db.query_user(user_id)  # 从数据库查
    r.setex(cache_key, 3600, json.dumps(user))  # 缓存1小时
    return user

RabbitMQ 消息队列

pip install pika

import pika, json

# 生产者（订单服务）
def publish_order_created(order_id, user_id):
    connection = pika.BlockingConnection(pika.ConnectionParameters('rabbitmq'))
    channel = connection.channel()
    channel.queue_declare(queue='order_created', durable=True)
    channel.basic_publish(
        exchange='',
        routing_key='order_created',
        body=json.dumps({'order_id': order_id, 'user_id': user_id}),
        properties=pika.BasicProperties(delivery_mode=2)  # 持久化
    )
    connection.close()

# 消费者（库存服务/通知服务）
def consume_order_events():
    connection = pika.BlockingConnection(pika.ConnectionParameters('rabbitmq'))
    channel = connection.channel()
    channel.queue_declare(queue='order_created', durable=True)

    def callback(ch, method, properties, body):
        event = json.loads(body)
        print(f"收到订单事件: {event}")
        process_order(event)  # 扣库存、发通知
        ch.basic_ack(delivery_tag=method.delivery_tag)

    channel.basic_qos(prefetch_count=1)
    channel.basic_consume(queue='order_created', on_message_callback=callback)
    print('等待订单消息...')
    channel.start_consuming()

Docker 部署

# order_service/Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["gunicorn", "-w", "4", "-b", "0.0.0.0:5000", "app:app"]

# docker-compose.yml
version: '3'
services:
  order-service:
    build: ./order_service
    ports: ["5002:5000"]
    depends_on: [redis, rabbitmq]
    environment:
      - REDIS_HOST=redis
      - RABBITMQ_HOST=rabbitmq

  user-service:
    build: ./user_service
    ports: ["5001:5000"]

  redis:
    image: redis:7-alpine
    ports: ["6379:6379"]

  rabbitmq:
    image: rabbitmq:3-management
    ports: ["5672:5672", "15672:15672"]

常见问题

Q1: 服务间调用失败怎么处理？用指数退避重试 + 熔断器模式（Hystrix/ pybreaker）。超过重试次数后返回降级响应。

Q2: 分布式事务怎么保证？用 Saga 模式（补偿事务）或 2PC（两阶段提交）。大多数微服务场景用最终一致性即可。

Q3: Docker 里的服务怎么互相发现？用 Docker Compose 的服务名作为 hostname（如 order-service 访问 user-service），生产环境用 Kubernetes Service 或 Consul。