事件驱动架构下微服务契约的版本兼容性与演进策略

摘要：本文探讨事件驱动架构下微服务间通过事件契约进行通信时的版本管理与演进策略。我们将构建一个模拟订单处理系统，演示如何利用Schema Registry（Apache Avro）实现契约的集中化、版本化存储与兼容性校验。核心内容包括向后兼容与向前兼容的设计原则、Schema演化规则（如添加可选字段、安全重命名），并通过一个完整可运行的Python项目（使用FastAPI、Confluent Kafka...

摘要

本文探讨事件驱动架构下微服务间通过事件契约进行通信时的版本管理与演进策略。我们将构建一个模拟订单处理系统，演示如何利用Schema Registry（Apache Avro）实现契约的集中化、版本化存储与兼容性校验。核心内容包括向后兼容与向前兼容的设计原则、Schema演化规则（如添加可选字段、安全重命名），并通过一个完整可运行的Python项目（使用FastAPI、Confluent Kafka）展示具体实现。项目将重点展示生产者如何发布版本化事件、消费者如何使用兼容性策略消费不同版本的事件，并提供清晰的演进示例与验证步骤。

1. 项目概述与设计思路

在事件驱动架构中，微服务通过发布和订阅事件进行异步通信。事件契约（Event Contract）定义了事件的格式（即Schema），是服务间交互的基石。随着业务发展，契约必然需要演进（如添加字段、修改类型）。若缺乏管理，不兼容的变更将导致消费者服务崩溃或数据丢失。

本项目旨在构建一个演示系统，解决以下核心问题：

契约集中化管理：使用Schema Registry存储Avro Schema，避免在各服务中硬编码。
版本兼容性保障：通过Schema Registry的兼容性检查（如BACKWARD模式），确保新注册的Schema与旧版本兼容。
运行时兼容性处理：消费者端能够消费其预期版本以及兼容的、不同版本的事件数据。
演进策略演示：通过具体代码示例，展示安全的演进操作（如添加字段）和破坏性变更的处理方法。

系统设计：我们模拟一个简化的订单处理流程。order-service接收创建订单的HTTP请求，随后发布OrderCreated事件。inventory-service和notification-service分别订阅该事件，进行库存扣减和发送通知（模拟）。所有事件均通过Kafka传递，并使用Avro序列化，Schema在Schema Registry中注册和查询。

2. 项目结构树

event-driven-avro-example/
├── docker-compose.yml
├── requirements.txt
├── config/
│   └── common_config.py
├── order_service/
│   ├── __init__.py
│   ├── main.py
│   ├── models.py
│   ├── producer.py
│   └── schemas/
│       ├── __init__.py
│       └── order_created_v1.avsc
│       └── order_created_v2.avsc
├── inventory_service/
│   ├── __init__.py
│   ├── main.py
│   └── consumer.py
├── notification_service/
│   ├── __init__.py
│   ├── main.py
│   └── consumer.py
└── scripts/
    └── test_events.py

3. 核心代码实现

文件路径：docker-compose.yml

version: '3.8'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-kafka:latest
    depends_on:

      - zookeeper
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    ports:

      - "9092:9092"

  schema-registry:
    image: confluentinc/cp-schema-registry:latest
    depends_on:

      - kafka
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: PLAINTEXT://kafka:9092
      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081
    ports:

      - "8081:8081"

  order-service:
    build: ./order_service
    depends_on:

      - kafka
      - schema-registry
    ports:

      - "8001:8000"
    volumes:

      - ./order_service:/app

  inventory-service:
    build: ./inventory_service
    depends_on:

      - kafka
      - schema-registry
    ports:

      - "8002:8000"
    volumes:

      - ./inventory_service:/app

  notification-service:
    build: ./notification_service
    depends_on:

      - kafka
      - schema-registry
    ports:

      - "8003:8000"
    volumes:

      - ./notification_service:/app

说明：定义了所有基础设施（Zookeeper, Kafka, Schema Registry）和三个微服务。每个服务使用独立的Dockerfile构建。

文件路径：config/common_config.py

# 通用配置
KAFKA_BOOTSTRAP_SERVERS = "kafka:9092"
SCHEMA_REGISTRY_URL = "http://schema-registry:8081"
ORDER_CREATED_TOPIC = "order.created"
# 主题名称与对应的Schema在Registry中的逻辑名称（通常称为Subject）
ORDER_CREATED_SUBJECT = f"{ORDER_CREATED_TOPIC}-value"

说明：集中管理配置，便于各服务引用。

文件路径：order_service/schemas/order_created_v1.avsc

{
  "type": "record",
  "name": "OrderCreated",
  "namespace": "com.example.events",
  "fields": [
    {"name": "order_id", "type": "string"},
    {"name": "customer_id", "type": "string"},
    {"name": "total_amount", "type": "double"},
    {"name": "items", "type": {"type": "array", "items": {
        "type": "record",
        "name": "OrderItem",
        "fields": [
          {"name": "product_id", "type": "string"},
          {"name": "quantity", "type": "int"},
          {"name": "price", "type": "double"}
        ]
      }}
    },
    {"name": "timestamp", "type": "long", "logicalType": "timestamp-millis"}
  ]
}

说明：事件契约的初始版本（v1），包含订单核心信息。

文件路径：order_service/schemas/order_created_v2.avsc

{
  "type": "record",
  "name": "OrderCreated",
  "namespace": "com.example.events",
  "fields": [
    {"name": "order_id", "type": "string"},
    {"name": "customer_id", "type": "string"},
    {"name": "total_amount", "type": "double"},
    {"name": "items", "type": {"type": "array", "items": {
        "type": "record",
        "name": "OrderItem",
        "fields": [
          {"name": "product_id", "type": "string"},
          {"name": "quantity", "type": "int"},
          {"name": "price", "type": "double"}
        ]
      }}
    },
    {"name": "timestamp", "type": "long", "logicalType": "timestamp-millis"},
    {"name": "customer_email", "type": ["null", "string"], "default": null}
  ]
}

说明：向后兼容的演进版本（v2），添加了一个可选字段customer_email（类型为[null, string]，默认值为null）。这是安全的"添加字段"操作。

文件路径：order_service/models.py

from pydantic import BaseModel
from typing import List
from datetime import datetime

# Pydantic模型用于接收HTTP请求
class OrderItemRequest(BaseModel):
    product_id: str
    quantity: int
    price: float

class OrderCreateRequest(BaseModel):
    customer_id: str
    items: List[OrderItemRequest]
    # V2新增字段，在HTTP API层面也是可选的，以演示演进
    customer_email: str = None

# 内部使用的订单模型
class Order(BaseModel):
    order_id: str
    customer_id: str
    customer_email: str = None  # 兼容V1，默认为None
    total_amount: float
    items: List[OrderItemRequest]
    timestamp: datetime

说明：定义了服务内部的请求和订单模型，与Avro Schema有对应关系，但独立存在。

文件路径：order_service/producer.py

from confluent_kafka import SerializingProducer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer
import logging
from config.common_config import KAFKA_BOOTSTRAP_SERVERS, SCHEMA_REGISTRY_URL, ORDER_CREATED_TOPIC, ORDER_CREATED_SUBJECT
import json

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class OrderEventProducer:
    def __init__(self):
        # 1. 连接Schema Registry
        schema_registry_conf = {'url': SCHEMA_REGISTRY_URL}
        schema_registry_client = SchemaRegistryClient(schema_registry_conf)
        
        # 2. 读取最新版本的Schema并创建Avro序列化器
        # 注意：在生产环境中，通常根据配置或环境变量决定使用的Schema版本
        # 此处为演示，我们注册并使用V2 Schema。注册操作通常在部署流程中完成。
        with open('schemas/order_created_v2.avsc', 'r') as f:
            schema_str = json.load(f) # Avro Schema是JSON格式
        # 创建序列化器，并指定兼容性检查（通常由Registry端配置，此处仅为示意）
        avro_serializer = AvroSerializer(schema_registry_client,
                                         schema_str,
                                         lambda obj, ctx: obj)
        
        # 3. 配置并创建Producer
        producer_conf = {
            'bootstrap.servers': KAFKA_BOOTSTRAP_SERVERS,
            'key.serializer': lambda key, ctx: str(key).encode('utf-8') if key else None,
            'value.serializer': avro_serializer
        }
        self.producer = SerializingProducer(producer_conf)
        self.topic = ORDER_CREATED_TOPIC
        logger.info("OrderEventProducer initialized.")

    def produce_order_created_event(self, order_data: dict):
        """
        发布OrderCreated事件。
        order_data 必须符合当前使用的Avro Schema结构。
        """
        def delivery_report(err, msg):
            if err is not None:
                logger.error(f'Message delivery failed: {err}')
            else:
                logger.info(f'Message delivered to {msg.topic()} [{msg.partition()}] @ offset {msg.offset()}')
        
        # 以order_id作为Kafka消息的Key，确保同一订单的事件有序
        self.producer.produce(topic=self.topic,
                              key=order_data['order_id'],
                              value=order_data,
                              on_delivery=delivery_report)
        self.producer.flush() # 同步发送，便于演示。生产环境可异步。
        logger.info(f"Produced OrderCreated event for order {order_data['order_id']}")

# 全局生产者实例
producer = OrderEventProducer()

说明：生产者负责将订单数据序列化为Avro格式并发布到Kafka。它依赖于Schema Registry来获取序列化所需的Schema。

文件路径：order_service/main.py

from fastapi import FastAPI, HTTPException
import uuid
from datetime import datetime
from .models import OrderCreateRequest, Order
from .producer import producer
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI(title="Order Service")

@app.post("/orders", response_model=Order)
async def create_order(order_req: OrderCreateRequest):
    # 1. 生成订单ID和计算总金额
    order_id = str(uuid.uuid4())
    total_amount = sum(item.price * item.quantity for item in order_req.items)
    
    # 2. 构建内部订单对象
    order = Order(
        order_id=order_id,
        customer_id=order_req.customer_id,
        customer_email=order_req.customer_email,
        total_amount=total_amount,
        items=order_req.items,
        timestamp=datetime.utcnow()
    )
    
    # 3. 构建符合Avro Schema的事件数据
    # 注意：需要将datetime转换为timestamp milliseconds
    event_data = order.dict()
    event_data['timestamp'] = int(order.timestamp.timestamp() * 1000)  # 转为毫秒时间戳
    # 处理items，确保是字典列表
    event_data['items'] = [item.dict() for item in order.items]
    
    # 4. 发布事件
    try:
        producer.produce_order_created_event(event_data)
    except Exception as e:
        logger.error(f"Failed to produce event: {e}")
        raise HTTPException(status_code=500, detail="Failed to process order")
    
    return order

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

sequenceDiagram participant Client participant OrderService participant SchemaRegistry participant Kafka participant InventoryService participant NotificationService Client->>OrderService: POST /orders (JSON) Note over OrderService: 生成订单ID，计算总额 OrderService->>SchemaRegistry: (隐式) 获取Schema用于序列化 SchemaRegistry-->>OrderService: Avro Schema (V2) OrderService->>OrderService: 构建符合Schema的事件数据 OrderService->>Kafka: Produce "order.created" (Avro, V2) Note over Kafka: 事件持久化 Kafka-->>InventoryService: 推送事件 InventoryService->>SchemaRegistry: 获取Schema用于反序列化 SchemaRegistry-->>InventoryService: Avro Schema (V2) InventoryService->>InventoryService: 消费事件，扣减库存 Kafka-->>NotificationService: 推送事件 NotificationService->>SchemaRegistry: 获取Schema用于反序列化 SchemaRegistry-->>NotificationService: Avro Schema (V2) NotificationService->>NotificationService: 消费事件，发送通知

图1：订单创建与事件消费的时序图，展示了服务间通过Schema Registry协调契约的交互过程。

文件路径：inventory_service/consumer.py

from confluent_kafka import DeserializingConsumer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroDeserializer
import logging
from config.common_config import KAFKA_BOOTSTRAP_SERVERS, SCHEMA_REGISTRY_URL, ORDER_CREATED_TOPIC
from typing import Dict, Any

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class InventoryEventConsumer:
    def __init__(self, group_id="inventory-service-group"):
        # 1. 连接Schema Registry
        schema_registry_conf = {'url': SCHEMA_REGISTRY_URL}
        schema_registry_client = SchemaRegistryClient(schema_registry_conf)
        
        # 2. 创建Avro反序列化器。
        # 关键：这里不指定具体Schema，反序列化器会从消息元数据中获取Schema ID，
        # 并从Registry中拉取对应的Schema来反序列化数据。
        # 这天然支持多版本消费，只要Registry中的Schema是兼容的。
        avro_deserializer = AvroDeserializer(schema_registry_client)
        
        # 3. 配置并创建Consumer
        consumer_conf = {
            'bootstrap.servers': KAFKA_BOOTSTRAP_SERVERS,
            'key.deserializer': lambda key, ctx: key.decode('utf-8') if key else None,
            'value.deserializer': avro_deserializer,
            'group.id': group_id,
            'auto.offset.reset': 'earliest',
            'enable.auto.commit': True
        }
        self.consumer = DeserializingConsumer(consumer_conf)
        self.topic = ORDER_CREATED_TOPIC
        logger.info("InventoryEventConsumer initialized.")

    def consume_and_process(self):
        self.consumer.subscribe([self.topic])
        logger.info(f"Subscribed to topic: {self.topic}")
        try:
            while True:
                msg = self.consumer.poll(1.0) # 超时时间1秒
                if msg is None:
                    continue
                if msg.error():
                    logger.error(f"Consumer error: {msg.error()}")
                    continue
                
                # 反序列化后的数据是一个Python字典
                order_event: Dict[str, Any] = msg.value()
                order_id = msg.key()
                
                # 处理事件 - 模拟库存扣减逻辑
                self._process_order_created(order_id, order_event)
                
        except KeyboardInterrupt:
            logger.info("Consumer interrupted.")
        finally:
            self.consumer.close()

    def _process_order_created(self, order_id: str, event_data: Dict[str, Any]):
        """处理OrderCreated事件的核心业务逻辑"""
        # 关键点：业务代码应能处理事件中可能缺失的字段（向前兼容）
        # 例如，如果此消费者代码是基于V1 Schema编写的，它不知道`customer_email`字段。
        # 但由于Avro反序列化和Schema兼容性，这个字段会被静默忽略或作为额外字段。
        # 我们应防御性地访问字段。
        customer_email = event_data.get('customer_email')
        # 即使event_data中没有'customer_email'键，`.get()`也会返回None，不会报错。
        
        items = event_data['items']
        for item in items:
            product_id = item['product_id']
            quantity = item['quantity']
            # 模拟库存扣减操作
            logger.info(f"[Inventory] Order {order_id}: Deducting {quantity} units for product {product_id}. Customer email in event: {customer_email}")
        
        # 模拟处理成功
        logger.info(f"[Inventory] Successfully processed order {order_id}")

# 全局消费者实例
consumer = InventoryEventConsumer()

文件路径：inventory_service/main.py

from .consumer import consumer
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def main():
    logger.info("Starting Inventory Service Consumer...")
    consumer.consume_and_process()

if __name__ == "__main__":
    main()

文件路径：notification_service/consumer.py

# notification_service/consumer.py 结构与inventory_service/consumer.py类似，但处理逻辑不同。
from confluent_kafka import DeserializingConsumer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroDeserializer
import logging
from config.common_config import KAFKA_BOOTSTRAP_SERVERS, SCHEMA_REGISTRY_URL, ORDER_CREATED_TOPIC

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class NotificationEventConsumer:
    def __init__(self, group_id="notification-service-group"):
        schema_registry_conf = {'url': SCHEMA_REGISTRY_URL}
        schema_registry_client = SchemaRegistryClient(schema_registry_conf)
        avro_deserializer = AvroDeserializer(schema_registry_client)
        
        consumer_conf = {
            'bootstrap.servers': KAFKA_BOOTSTRAP_SERVERS,
            'key.deserializer': lambda key, ctx: key.decode('utf-8') if key else None,
            'value.deserializer': avro_deserializer,
            'group.id': group_id,
            'auto.offset.reset': 'earliest'
        }
        self.consumer = DeserializingConsumer(consumer_conf)
        self.topic = ORDER_CREATED_TOPIC
        logger.info("NotificationEventConsumer initialized.")

    def consume_and_process(self):
        self.consumer.subscribe([self.topic])
        logger.info(f"Subscribed to topic: {self.topic}")
        try:
            while True:
                msg = self.consumer.poll(1.0)
                if msg is None:
                    continue
                if msg.error():
                    logger.error(f"Consumer error: {msg.error()}")
                    continue
                
                order_event = msg.value()
                order_id = msg.key()
                self._process_order_created(order_id, order_event)
                
        except KeyboardInterrupt:
            logger.info("Consumer interrupted.")
        finally:
            self.consumer.close()

    def _process_order_created(self, order_id: str, event_data: dict):
        """处理OrderCreated事件，发送通知"""
        # 此服务需要customer_email字段，它是基于V2契约编写的。
        # 如果消费到V1事件（无此字段），我们使用备用方案。
        customer_email = event_data.get('customer_email')
        customer_id = event_data['customer_id']
        
        if customer_email:
            # 优先使用邮箱通知
            logger.info(f"[Notification] Order {order_id}: Sending email to {customer_email}")
        else:
            # 回退方案：记录日志或通过其他渠道通知（如站内信，关联customer_id）
            logger.info(f"[Notification] Order {order_id}: No email provided for customer {customer_id}. Sending in-app notification.")
        
        logger.info(f"[Notification] Successfully processed notification for order {order_id}")

consumer = NotificationEventConsumer()

文件路径：notification_service/main.py

from .consumer import consumer
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def main():
    logger.info("Starting Notification Service Consumer...")
    consumer.consume_and_process()

if __name__ == "__main__":
    main()

图2：系统组件与数据流图，展示了事件从产生、注册、传递到消费的完整路径，以及各组件角色。

文件路径：scripts/test_events.py

#!/usr/bin/env python3
"""
测试脚本：模拟客户端调用，并验证事件流转。
"""
import requests
import json
import time
import logging
from confluent_kafka import Consumer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroDeserializer

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

ORDER_SERVICE_URL = "http://localhost:8001/orders"
KAFKA_BOOTSTRAP_SERVERS = "localhost:9092"
SCHEMA_REGISTRY_URL = "http://localhost:8081"
TOPIC = "order.created"

def test_order_creation_v1():
    """测试发送一个符合V1 Schema的订单（不包含customer_email）"""
    payload = {
        "customer_id": "cust_123",
        "items": [
            {"product_id": "prod_001", "quantity": 2, "price": 25.5},
            {"product_id": "prod_002", "quantity": 1, "price": 99.99}
        ]
    }
    logger.info("Testing order creation (V1 compatible, no email)...")
    resp = requests.post(ORDER_SERVICE_URL, json=payload)
    logger.info(f"Response: {resp.status_code}, {resp.json()}")
    return resp.json().get('order_id')

def test_order_creation_v2():
    """测试发送一个符合V2 Schema的订单（包含customer_email）"""
    payload = {
        "customer_id": "cust_456",
        "customer_email": "user@example.com",
        "items": [
            {"product_id": "prod_003", "quantity": 5, "price": 10.0}
        ]
    }
    logger.info("Testing order creation (V2 with email)...")
    resp = requests.post(ORDER_SERVICE_URL, json=payload)
    logger.info(f"Response: {resp.status_code}, {resp.json()}")
    return resp.json().get('order_id')

def consume_and_verify_last_event(expected_order_id=None):
    """创建一个临时消费者，拉取最新的消息并验证其Key"""
    schema_registry_client = SchemaRegistryClient({'url': SCHEMA_REGISTRY_URL})
    avro_deserializer = AvroDeserializer(schema_registry_client)
    
    consumer_conf = {
        'bootstrap.servers': KAFKA_BOOTSTRAP_SERVERS,
        'group.id': 'test-verification-group',
        'auto.offset.reset': 'latest', # 从最新的开始，只读新消息
        'enable.auto.commit': False,
        'session.timeout.ms': 6000,
    }
    consumer = Consumer(consumer_conf)
    consumer.subscribe([TOPIC])
    
    logger.info("Polling for new event...")
    start_time = time.time()
    timeout = 10
    while time.time() - start_time < timeout:
        msg = consumer.poll(1.0)
        if msg is None:
            continue
        if msg.error():
            logger.error(f"Error: {msg.error()}")
            break
        
        # 反序列化值
        value = avro_deserializer(msg.value(), None)
        key = msg.key().decode('utf-8')
        logger.info(f"Consumed event -> Key: {key}, Value (summary): { {k: type(v).__name__ for k, v in value.items()} }")
        
        if expected_order_id and key == expected_order_id:
            logger.info(f"✓ Successfully verified event for order {expected_order_id}")
            consumer.close()
            return True
    consumer.close()
    logger.warning("Timeout waiting for event.")
    return False

if __name__ == "__main__":
    logger.info("Starting integration test...")
    # 等待服务启动
    time.sleep(5)
    
    # 测试V1兼容性
    order_id_1 = test_order_creation_v1()
    time.sleep(2)
    consume_and_verify_last_event(order_id_1)
    
    # 测试V2
    order_id_2 = test_order_creation_v2()
    time.sleep(2)
    consume_and_verify_last_event(order_id_2)
    
    logger.info("Integration test finished. Check logs of inventory and notification services for processing details.")

文件路径：order_service/Dockerfile & 其他服务Dockerfile

# order_service/Dockerfile (其他服务类似，仅WORKDIR和启动命令不同)
FROM python:3.9-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["python", "main.py"]

文件路径：requirements.txt

fastapi==0.104.1
uvicorn[standard]==0.24.0
confluent-kafka==2.2.0
avro==1.11.3
pydantic==2.5.0
requests==2.31.0

4. 安装依赖与运行步骤

mkdir event-driven-avro-example && cd event-driven-avro-example
    # 将上述所有文件和目录按照结构树放置。

启动基础设施：

# 在项目根目录下运行，启动Kafka, Zookeeper, Schema Registry
    docker-compose up -d zookeeper kafka schema-registry
    # 等待几十秒确保服务就绪
    sleep 30

构建并启动微服务：

# 构建三个服务的镜像并启动
    docker-compose up -d order-service inventory-service notification-service
    # 查看日志，确保服务启动无报错
    docker-compose logs -f order-service
    # 在其他终端查看消费者日志
    # docker-compose logs -f inventory-service
    # docker-compose logs -f notification-service

5. 测试与验证步骤

手动测试API：

# 创建V1风格订单（无邮箱）
    curl -X POST "http://localhost:8001/orders" \
      -H "Content-Type: application/json" \
      -d '{
        "customer_id": "test_customer_1",
        "items": [
          {"product_id": "book-123", "quantity": 1, "price": 29.99}
        ]
      }'
    
    # 创建V2风格订单（有邮箱）
    curl -X POST "http://localhost:8001/orders" \
      -H "Content-Type: application/json" \
      -d '{
        "customer_id": "test_customer_2",
        "customer_email": "alice@example.com",
        "items": [
          {"product_id": "ebook-456", "quantity": 1, "price": 15.50}
        ]
      }'

观察`inventory-service`和`notification-service`的日志输出，两者都应成功处理两个订单。对于第一个订单，`notification-service`会回退到站内通知。

运行集成测试脚本：

# 在宿主机上安装依赖（或在虚拟环境中）
    pip install -r requirements.txt
    # 运行测试脚本，它会调用API并验证Kafka中的事件
    python scripts/test_events.py

验证Schema注册与兼容性：

# 查看已注册的Subjects
    curl http://localhost:8081/subjects
    # 应返回 ["order.created-value"]
    
    # 查看Subject的所有版本
    curl http://localhost:8081/subjects/order.created-value/versions
    # 可能返回 [1, 2]
    
    # 查看特定版本的Schema
    curl http://localhost:8081/subjects/order.created-value/versions/1 | jq .
    curl http://localhost:8081/subjects/order.created-value/versions/2 | jq .
    
    # 测试兼容性（尝试将V1 Schema作为新版本注册，应失败，因为V2添加了字段，默认是BACKWARD兼容）
    # BACKWARD: 新Schema（作为消费者）可以读取旧数据。即V2可以读V1数据。
    # 我们测试FORWARD兼容性（可选）：
    # curl -X POST -H "Content-Type: application/json" \
    #   --data '{"schema": "{\"type\":\"record\",\"name\":\"OrderCreated\",\"namespace\":\"com.example.events\",\"fields\":[{\"name\":\"order_id\",\"type\":\"string\"},{\"name\":\"customer_id\",\"type\":\"string\"}]}"}' \
    #   "http://localhost:8081/compatibility/subjects/order.created-value/versions/latest"

6. 演进策略与扩展说明

本项目演示了向后兼容的演进（添加可选字段）。其他常见策略包括：

向前兼容：旧消费者能忽略新字段。本项目中的消费者使用.get()方法访问字段，实现了这一点。
安全的重命名：通过别名（Alias）功能，可以在Schema Registry和Avro反序列化器的支持下，将字段old_name重命名为new_name，同时保持双向兼容。
破坏性变更处理：对于必须进行的破坏性变更（如删除必需字段、修改字段类型），策略包括：
1. 双写/双发：在一段时间内，同时发布新旧两个版本的事件到不同主题。
2. 消费者升级窗口：先升级所有消费者，使其能处理新版本事件，然后升级生产者。
3. 使用事件信封（Envelope）：在事件外层包装元数据，内部承载不同版本的数据体。

生产建议：

将Schema的注册与兼容性检查集成到CI/CD流水线中。
为每个事件主题配置明确的兼容性类型（如BACKWARD、BACKWARD_TRANSITIVE）。
监控Schema Registry，跟踪Schema版本变化。
消费者服务应记录其消费的事件Schema版本，便于排查问题。

通过本项目的代码与实践，可以建立起事件驱动架构下稳健的契约演进机制，保障分布式系统在持续迭代中的稳定性与弹性。