第6章：微服务架构¶

# 用户服务使用MySQL
class UserService:
    def __init__(self):
        self.db = MySQLConnection('user_db')

    def get_user(self, user_id):
        return self.db.query("SELECT * FROM users WHERE id = %s", user_id)

# 订单服务使用MongoDB
class OrderService:
    def __init__(self):
        self.db = MongoClient('order_db')

    def get_order(self, order_id):
        return self.db.orders.find_one({'id': order_id})

# 示例：从单体应用拆分微服务

# 原单体应用
class MonolithicApp:
    def create_order(self, user_id, product_id):
        # 获取用户
        user = self.get_user(user_id)
        # 获取商品
        product = self.get_product(product_id)
        # 创建订单
        order = self.create_order_record(user, product)
        return order

# 拆分后的微服务
class OrderService:
    def __init__(self):
        self.user_client = UserClient()
        self.product_client = ProductClient()
        self.db = OrderDatabase()

    def create_order(self, user_id, product_id):
        # 调用用户服务
        user = self.user_client.get_user(user_id)
        # 调用商品服务
        product = self.product_client.get_product(product_id)
        # 创建订单
        order = self.db.create_order(user, product)
        return order

import consul

# 服务注册
consul_client = consul.Consul(host='localhost', port=8500)

consul_client.agent.service.register(
    name='user-service',
    service_id='user-service-1',
    address='localhost',
    port=8001,
    tags=['user', 'api']
)

# 服务发现
services = consul_client.agent.services()
user_services = [s for s in services.values() if s['Service'] == 'user-service']

# 调用服务
if user_services:
    service = user_services[0]
    url = f"http://{service['Address']}:{service['Port']}/user/1"
    response = requests.get(url)

from nacos import NacosClient

# 服务注册
client = NacosClient('localhost', 8848)

client.add_naming_instance(
    service_name='user-service',
    ip='127.0.0.1',
    port=8001,
    cluster_name='DEFAULT',
    weight=1.0
)

# 服务发现
instances = client.list_naming_instance(
    service_name='user-service',
    cluster_name='DEFAULT'
)

# 调用服务
if instances:
    instance = instances['hosts'][0]
    url = f"http://{instance['ip']}:{instance['port']}/user/1"
    response = requests.get(url)

from apollo_client import Config

# 创建配置客户端
config = Config(
    app_id='user-service',
    namespace='application',
    config_server_url='http://localhost:8080'
)

# 获取配置
app_name = config.get_value('app.name', 'default')
app_version = config.get_value('app.version', '1.0.0')

# 监听配置变更
def config_change_handler(change):
    print(f"Config changed: {change}")

config.add_change_listener(config_change_handler)

from fastapi import FastAPI, HTTPException
import httpx

app = FastAPI()

# 用户服务
@app.get("/users/{user_id}")
async def get_user(user_id: int):  # async def定义异步函数；用await调用
    user = db.query("SELECT * FROM users WHERE id = %s", user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

# 订单服务调用用户服务
@app.post("/orders")
async def create_order(user_id: int, product_id: int):
    # 调用用户服务
    async with httpx.AsyncClient() as client:
        response = await client.get(f"http://user-service/users/{user_id}")  # await等待异步操作完成
        if response.status_code == 404:
            raise HTTPException(status_code=404, detail="User not found")
        user = response.json()

    # 创建订单
    order = create_order_record(user, product_id)
    return order

# 服务端实现
import grpc
from concurrent import futures

class UserServiceImpl(user_service_pb2_grpc.UserServiceServicer):
    def GetUser(self, request, context):
        user = db.query("SELECT * FROM users WHERE id = %s", request.user_id)
        if not user:
            context.set_code(grpc.StatusCode.NOT_FOUND)
            context.set_details('User not found')
            return user_service_pb2.GetUserResponse()

        return user_service_pb2.GetUserResponse(
            id=user['id'],
            name=user['name'],
            email=user['email']
        )

# 客户端调用
# ⚠️ 安全警告：grpc.insecure_channel() 不使用 TLS 加密，数据以明文传输。
# 生产环境请使用 grpc.secure_channel() 并配置 TLS 证书：
#   credentials = grpc.ssl_channel_credentials(root_certificates=...)
#   channel = grpc.secure_channel('host:port', credentials)
def get_user(user_id):
    with grpc.insecure_channel('localhost:50051') as channel:
        stub = user_service_pb2_grpc.UserServiceStub(channel)
        response = stub.GetUser(user_service_pb2.GetUserRequest(user_id=user_id))
        return response

# 使用消息队列进行异步通信
import json
from kafka import KafkaProducer, KafkaConsumer

# 生产者：订单服务
producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')  # lambda匿名函数：简洁的单行函数
)

def create_order(user_id, product_id):
    # 创建订单
    order = db.create_order(user_id, product_id)

    # 发送消息到库存服务
    producer.send('order-created', {
        'order_id': order['id'],
        'product_id': product_id,
        'quantity': 1
    })

    return order

# 消费者：库存服务
consumer = KafkaConsumer(
    'order-created',
    bootstrap_servers=['localhost:9092'],
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))  # json.loads将JSON字符串转为Python对象
)

for message in consumer:
    # 扣减库存
    deduct_inventory(message.value['product_id'], message.value['quantity'])