【系统架构设计百科】单体到微服务：迁移策略与 Strangler Fig 模式

# Spring Cloud Gateway 配置示例
spring:
  cloud:
    gateway:
      routes:
        - id: order-service-route
          uri: lb://order-service
          predicates:
            - Path=/api/v1/orders/**
          filters:
            - StripPrefix=0
            - name: CircuitBreaker
              args:
                name: orderServiceCB
                fallbackUri: forward:/fallback/monolith/orders

        - id: user-service-route
          uri: lb://user-service
          predicates:
            - Path=/api/v1/users/**

        # 兜底路由：所有未匹配的请求转发到单体
        - id: monolith-fallback
          uri: http://monolith.internal:8080
          predicates:
            - Path=/**
          order: 9999

@Component
public class StranglerRouter {

    @Value("${migration.orders.enabled:false}")
    private boolean ordersServiceEnabled;

    @Value("${migration.orders.percentage:0}")
    private int ordersTrafficPercentage;

    private final MonolithOrderService monolithService;
    private final MicroserviceOrderClient microserviceClient;
    private final Random random = new Random();

    public StranglerRouter(MonolithOrderService monolithService,
                           MicroserviceOrderClient microserviceClient) {
        this.monolithService = monolithService;
        this.microserviceClient = microserviceClient;
    }

    public Order getOrder(String orderId) {
        if (ordersServiceEnabled && shouldRouteToNewService()) {
            try {
                return microserviceClient.getOrder(orderId);
            } catch (ServiceUnavailableException e) {
                // 降级回退到单体
                log.warn("微服务不可用，回退到单体: {}", e.getMessage());
                return monolithService.getOrder(orderId);
            }
        }
        return monolithService.getOrder(orderId);
    }

    private boolean shouldRouteToNewService() {
        return random.nextInt(100) < ordersTrafficPercentage;
    }
}

# 领域分类示例（用于辅助决策的分析脚本）
from dataclasses import dataclass
from enum import Enum
from typing import List

class DomainType(Enum):
    CORE = "核心域"        # 业务核心竞争力
    SUPPORTING = "支撑域"   # 支撑核心业务，但非差异化竞争力
    GENERIC = "通用域"      # 行业通用，可外购

@dataclass
class BoundedContext:
    name: str
    domain_type: DomainType
    business_capability: str
    data_entities: List[str]
    upstream_dependencies: List[str]
    downstream_consumers: List[str]
    estimated_team_size: int
    change_frequency: str  # high / medium / low

# 电商系统的限界上下文分析
contexts = [
    BoundedContext(
        name="订单管理",
        domain_type=DomainType.CORE,
        business_capability="处理用户下单、支付、退款全流程",
        data_entities=["Order", "OrderItem", "Payment", "Refund"],
        upstream_dependencies=["商品目录", "库存", "定价"],
        downstream_consumers=["物流", "通知", "数据分析"],
        estimated_team_size=5,
        change_frequency="high"
    ),
    BoundedContext(
        name="库存管理",
        domain_type=DomainType.SUPPORTING,
        business_capability="管理商品库存数量和仓库分配",
        data_entities=["Stock", "Warehouse", "StockMovement"],
        upstream_dependencies=["商品目录"],
        downstream_consumers=["订单管理", "采购"],
        estimated_team_size=3,
        change_frequency="medium"
    ),
    BoundedContext(
        name="用户认证",
        domain_type=DomainType.GENERIC,
        business_capability="用户注册、登录、权限管理",
        data_entities=["User", "Role", "Permission", "Session"],
        upstream_dependencies=[],
        downstream_consumers=["订单管理", "商品目录", "通知"],
        estimated_team_size=2,
        change_frequency="low"
    ),
]

def analyze_migration_priority(contexts: List[BoundedContext]):
    """根据变更频率和依赖关系分析迁移优先级"""
    priority_score = {}
    for ctx in contexts:
        score = 0
        # 高变更频率的上下文优先迁移
        if ctx.change_frequency == "high":
            score += 3
        elif ctx.change_frequency == "medium":
            score += 2
        else:
            score += 1
        # 下游消费者少的优先迁移（依赖影响小）
        score += max(0, 5 - len(ctx.downstream_consumers))
        # 上游依赖少的优先迁移（自包含性高）
        score += max(0, 5 - len(ctx.upstream_dependencies))
        priority_score[ctx.name] = score

    sorted_priorities = sorted(
        priority_score.items(), key=lambda x: x[1], reverse=True
    )
    return sorted_priorities

priorities = analyze_migration_priority(contexts)
for name, score in priorities:
    print(f"迁移优先级 - {name}: {score} 分")

/**
 * 防腐层示例：将单体的订单模型转换为微服务的订单模型
 */
@Component
public class OrderAntiCorruptionLayer {

    private final MonolithOrderRepository monolithRepo;

    public OrderAntiCorruptionLayer(MonolithOrderRepository monolithRepo) {
        this.monolithRepo = monolithRepo;
    }

    /**
     * 将单体的订单数据转换为新服务的领域模型
     */
    public OrderAggregate translateFromMonolith(String orderId) {
        // 从单体数据库读取原始数据
        MonolithOrderDTO raw = monolithRepo.findById(orderId);

        // 转换为新服务的领域模型
        OrderAggregate order = OrderAggregate.builder()
            .id(OrderId.of(raw.getOrderNo()))
            .customerId(CustomerId.of(raw.getUserId()))
            .status(mapStatus(raw.getStatusCode()))
            .items(translateItems(raw.getOrderLines()))
            .shippingAddress(translateAddress(raw.getDeliveryAddr()))
            .createdAt(raw.getCreateTime().toInstant())
            .build();

        return order;
    }

    private OrderStatus mapStatus(int legacyCode) {
        return switch (legacyCode) {
            case 0 -> OrderStatus.DRAFT;
            case 1 -> OrderStatus.PENDING_PAYMENT;
            case 2 -> OrderStatus.PAID;
            case 3 -> OrderStatus.SHIPPED;
            case 4 -> OrderStatus.DELIVERED;
            case 5 -> OrderStatus.CANCELLED;
            case 9 -> OrderStatus.REFUNDED;
            default -> {
                log.warn("未知的旧系统状态码: {}，映射为 UNKNOWN", legacyCode);
                yield OrderStatus.UNKNOWN;
            }
        };
    }

    private List<OrderItem> translateItems(List<MonolithOrderLineDTO> lines) {
        return lines.stream()
            .map(line -> OrderItem.builder()
                .productId(ProductId.of(line.getSkuCode()))
                .productName(line.getItemName())
                .quantity(Quantity.of(line.getQty()))
                .unitPrice(Money.of(line.getPrice(), Currency.CNY))
                .build())
            .toList();
    }

    private ShippingAddress translateAddress(String rawAddress) {
        // 旧系统将地址存储为单个字段，用竖线分隔
        String[] parts = rawAddress.split("\\|");
        return ShippingAddress.builder()
            .province(parts.length > 0 ? parts[0] : "")
            .city(parts.length > 1 ? parts[1] : "")
            .district(parts.length > 2 ? parts[2] : "")
            .street(parts.length > 3 ? parts[3] : "")
            .build();
    }
}

-- 阶段一：为新服务创建独立的数据库用户和视图
CREATE USER 'order_service'@'%' IDENTIFIED BY '${ORDER_SVC_PASSWORD}';

-- 只授权新服务需要的表
GRANT SELECT, INSERT, UPDATE ON ecommerce.orders TO 'order_service'@'%';
GRANT SELECT, INSERT, UPDATE ON ecommerce.order_items TO 'order_service'@'%';
GRANT SELECT ON ecommerce.products TO 'order_service'@'%';

-- 创建视图隐藏不需要的列
CREATE VIEW order_service_products AS
SELECT
    product_id,
    product_name,
    sku_code,
    current_price
FROM ecommerce.products
WHERE is_active = 1;

# Debezium CDC 连接器配置
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: monolith-order-cdc
  labels:
    strimzi.io/cluster: debezium-connect
spec:
  class: io.debezium.connector.mysql.MySqlConnector
  tasksMax: 1
  config:
    database.hostname: monolith-db.internal
    database.port: 3306
    database.user: cdc_reader
    database.password: ${CDC_PASSWORD}
    database.server.id: 100001
    database.server.name: monolith
    database.include.list: ecommerce
    table.include.list: ecommerce.orders,ecommerce.order_items
    database.history.kafka.bootstrap.servers: kafka:9092
    database.history.kafka.topic: schema-changes.monolith
    transforms: route
    transforms.route.type: org.apache.kafka.connect.transforms.RegexRouter
    transforms.route.regex: "([^.]+)\\.([^.]+)\\.([^.]+)"
    transforms.route.replacement: "cdc.$3"
    snapshot.mode: initial
    decimal.handling.mode: double
    time.precision.mode: connect

@KafkaListener(topics = "cdc.orders", groupId = "order-service-sync")
public class OrderCdcConsumer {

    private final OrderRepository orderRepository;
    private final OrderAntiCorruptionLayer acl;

    public OrderCdcConsumer(OrderRepository orderRepository,
                            OrderAntiCorruptionLayer acl) {
        this.orderRepository = orderRepository;
        this.acl = acl;
    }

    @KafkaHandler
    public void handleOrderChange(ConsumerRecord<String, String> record) {
        CdcEvent event = CdcEvent.parse(record.value());

        switch (event.getOperation()) {
            case CREATE, UPDATE -> {
                OrderAggregate order = acl.translateFromCdcPayload(
                    event.getAfter()
                );
                orderRepository.save(order);
                log.info("同步订单数据: orderId={}, op={}",
                    order.getId(), event.getOperation());
            }
            case DELETE -> {
                String orderId = event.getBefore().get("order_id").asText();
                orderRepository.softDelete(OrderId.of(orderId));
                log.info("标记订单删除: orderId={}", orderId);
            }
        }
    }
}

@Component
public class CreateOrderSaga {

    private final SagaOrchestrator orchestrator;

    public CreateOrderSaga(SagaOrchestrator orchestrator) {
        this.orchestrator = orchestrator;
    }

    public SagaDefinition<CreateOrderSagaData> buildSaga() {
        return orchestrator
            .step("验证库存")
                .invoke(this::reserveInventory)
                .compensate(this::releaseInventory)
            .step("扣款")
                .invoke(this::processPayment)
                .compensate(this::refundPayment)
            .step("创建订单")
                .invoke(this::createOrder)
                .compensate(this::cancelOrder)
            .step("通知物流")
                .invoke(this::notifyShipping)
            .build();
    }

    private void reserveInventory(CreateOrderSagaData data) {
        inventoryClient.reserve(
            data.getItems().stream()
                .map(item -> new ReserveRequest(
                    item.getProductId(),
                    item.getQuantity()))
                .toList(),
            data.getSagaId()
        );
    }

    private void releaseInventory(CreateOrderSagaData data) {
        inventoryClient.release(data.getSagaId());
    }

    private void processPayment(CreateOrderSagaData data) {
        paymentClient.charge(
            data.getCustomerId(),
            data.getTotalAmount(),
            data.getSagaId()
        );
    }

    private void refundPayment(CreateOrderSagaData data) {
        paymentClient.refund(data.getSagaId());
    }

    private void createOrder(CreateOrderSagaData data) {
        orderRepository.save(data.toOrderAggregate());
    }

    private void cancelOrder(CreateOrderSagaData data) {
        orderRepository.updateStatus(
            data.getOrderId(), OrderStatus.CANCELLED
        );
    }

    private void notifyShipping(CreateOrderSagaData data) {
        shippingClient.schedulePickup(data.getOrderId());
    }
}

-- 在业务数据库中创建发件箱表
CREATE TABLE outbox_events (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    aggregate_type VARCHAR(100) NOT NULL,
    aggregate_id VARCHAR(100) NOT NULL,
    event_type VARCHAR(200) NOT NULL,
    payload JSON NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    published_at TIMESTAMP NULL,
    INDEX idx_unpublished (published_at) 
);

-- 业务操作和事件发布在同一事务中
START TRANSACTION;

INSERT INTO orders (order_id, customer_id, status, total_amount)
VALUES ('ORD-20260413-001', 'CUST-1001', 'CREATED', 599.00);

INSERT INTO outbox_events (aggregate_type, aggregate_id, event_type, payload)
VALUES (
    'Order',
    'ORD-20260413-001',
    'OrderCreated',
    '{"orderId":"ORD-20260413-001","customerId":"CUST-1001","totalAmount":599.00}'
);

COMMIT;

import hashlib
import json
from datetime import datetime, timedelta

class DualWriteVerifier:
    """双写校验器：比较新旧数据库中的数据是否一致"""

    def __init__(self, legacy_db, new_db, alert_service):
        self.legacy_db = legacy_db
        self.new_db = new_db
        self.alert_service = alert_service

    def verify_orders(self, time_window_minutes=10):
        """校验指定时间窗口内的订单数据一致性"""
        cutoff = datetime.utcnow() - timedelta(minutes=time_window_minutes)

        legacy_orders = self.legacy_db.execute(
            "SELECT order_id, customer_id, status, total_amount, "
            "updated_at FROM orders WHERE updated_at >= %s",
            (cutoff,)
        )

        inconsistencies = []
        for legacy in legacy_orders:
            new_order = self.new_db.execute(
                "SELECT order_id, customer_id, status, total_amount, "
                "updated_at FROM orders WHERE order_id = %s",
                (legacy["order_id"],)
            )

            if not new_order:
                inconsistencies.append({
                    "type": "MISSING_IN_NEW",
                    "order_id": legacy["order_id"],
                    "legacy_data": legacy
                })
                continue

            new = new_order[0]
            legacy_hash = self._compute_hash(legacy)
            new_hash = self._compute_hash(new)

            if legacy_hash != new_hash:
                inconsistencies.append({
                    "type": "DATA_MISMATCH",
                    "order_id": legacy["order_id"],
                    "legacy_data": legacy,
                    "new_data": new,
                    "diff_fields": self._find_diff_fields(legacy, new)
                })

        if inconsistencies:
            self.alert_service.send_alert(
                severity="HIGH",
                title=f"双写校验发现 {len(inconsistencies)} 条不一致数据",
                details=json.dumps(inconsistencies, default=str)
            )

        return inconsistencies

    def _compute_hash(self, record):
        normalized = json.dumps(
            {k: str(v) for k, v in sorted(record.items())
             if k != "updated_at"},
            sort_keys=True
        )
        return hashlib.sha256(normalized.encode()).hexdigest()

    def _find_diff_fields(self, legacy, new):
        return [
            k for k in legacy.keys()
            if k != "updated_at" and str(legacy.get(k)) != str(new.get(k))
        ]

# Istio VirtualService 配置示例
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: order-service
spec:
  hosts:
    - order-api.example.com
  http:
    - match:
        - uri:
            prefix: /api/v1/orders
      route:
        # 90% 流量到单体
        - destination:
            host: monolith.default.svc.cluster.local
            port:
              number: 8080
          weight: 90
        # 10% 流量到新服务
        - destination:
            host: order-service.default.svc.cluster.local
            port:
              number: 8080
          weight: 10
      retries:
        attempts: 3
        perTryTimeout: 2s
      timeout: 10s

@Aspect
@Component
public class DarkLaunchAspect {

    private final MeterRegistry meterRegistry;
    private final ExecutorService darkLaunchExecutor;

    public DarkLaunchAspect(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
        this.darkLaunchExecutor = Executors.newFixedThreadPool(
            10, new NamedThreadFactory("dark-launch")
        );
    }

    @Around("@annotation(darkLaunch)")
    public Object executeDarkLaunch(ProceedingJoinPoint joinPoint,
                                     DarkLaunch darkLaunch) throws Throwable {
        // 执行旧系统的处理逻辑，返回旧系统的结果
        Object legacyResult = joinPoint.proceed();

        // 异步调用新服务进行对比
        darkLaunchExecutor.submit(() -> {
            try {
                long start = System.nanoTime();
                Object newResult = invokeNewService(
                    darkLaunch.newServiceMethod(),
                    joinPoint.getArgs()
                );
                long duration = System.nanoTime() - start;

                // 记录性能指标
                meterRegistry.timer("dark_launch.latency",
                    "service", darkLaunch.serviceName()
                ).record(duration, TimeUnit.NANOSECONDS);

                // 对比结果
                boolean match = compareResults(legacyResult, newResult);
                meterRegistry.counter("dark_launch.comparison",
                    "service", darkLaunch.serviceName(),
                    "match", String.valueOf(match)
                ).increment();

                if (!match) {
                    log.warn("暗启动结果不匹配: method={}, legacy={}, new={}",
                        darkLaunch.newServiceMethod(),
                        legacyResult, newResult);
                }
            } catch (Exception e) {
                meterRegistry.counter("dark_launch.errors",
                    "service", darkLaunch.serviceName()
                ).increment();
                log.error("暗启动调用失败", e);
            }
        });

        return legacyResult;
    }
}

@RestController
@RequestMapping("/api/v1/orders")
public class OrderController {

    private final FeatureToggleService featureToggle;
    private final LegacyOrderService legacyService;
    private final NewOrderService newService;

    @GetMapping("/{orderId}")
    public ResponseEntity<OrderDTO> getOrder(
            @PathVariable String orderId,
            @RequestHeader("X-User-Id") String userId) {

        if (featureToggle.isEnabled("order.read.new-service", userId)) {
            try {
                OrderDTO result = newService.getOrder(orderId);
                return ResponseEntity.ok(result);
            } catch (Exception e) {
                log.warn("新服务异常，降级到旧系统: {}", e.getMessage());
                if (featureToggle.isEnabled("order.read.fallback-enabled")) {
                    return ResponseEntity.ok(
                        legacyService.getOrder(orderId)
                    );
                }
                throw e;
            }
        }

        return ResponseEntity.ok(legacyService.getOrder(orderId));
    }
}

# 使用 Prometheus 和 Grafana 建立基线监控
# prometheus.yml 配置片段
cat <<'EOF'
scrape_configs:
  - job_name: 'monolith'
    metrics_path: '/actuator/prometheus'
    static_configs:
      - targets: ['monolith.internal:8080']
    scrape_interval: 15s

  - job_name: 'order-service'
    metrics_path: '/actuator/prometheus'
    static_configs:
      - targets: ['order-service.internal:8080']
    scrape_interval: 15s
EOF

# 需要采集的基线指标清单
baseline_metrics:
  performance:
    - name: request_latency_p50
      description: "请求延迟中位数"
      target: "< 100ms"
    - name: request_latency_p99
      description: "请求延迟 P99"
      target: "< 500ms"
    - name: throughput_rps
      description: "每秒请求数"
      target: "> 1000"

  reliability:
    - name: error_rate
      description: "错误率（5xx 占比）"
      target: "< 0.1%"
    - name: availability
      description: "可用性（成功请求占比）"
      target: "> 99.95%"

  business:
    - name: order_completion_rate
      description: "订单完成率"
      target: "> 98%"
    - name: payment_success_rate
      description: "支付成功率"
      target: "> 99.5%"

# GitHub Actions 工作流示例
name: Order Service CI/CD

on:
  push:
    branches: [main]
    paths:
      - 'services/order-service/**'
  pull_request:
    branches: [main]
    paths:
      - 'services/order-service/**'

jobs:
  build-and-test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: 设置 JDK 21
        uses: actions/setup-java@v4
        with:
          java-version: '21'
          distribution: 'temurin'

      - name: 构建与单元测试
        run: |
          cd services/order-service
          ./gradlew build test

      - name: 集成测试（含单体兼容性验证）
        run: |
          cd services/order-service
          ./gradlew integrationTest \
            -Dmonolith.url=http://monolith-staging.internal:8080 \
            -Dverify.dual-write=true

      - name: 契约测试
        run: |
          cd services/order-service
          ./gradlew contractTest

      - name: 构建 Docker 镜像
        if: github.ref == 'refs/heads/main'
        run: |
          cd services/order-service
          docker build -t order-service:${{ github.sha }} .

      - name: 部署到预发环境
        if: github.ref == 'refs/heads/main'
        run: |
          kubectl set image deployment/order-service \
            order-service=order-service:${{ github.sha }} \
            -n staging

@SpringBootTest
@AutoConfigureMockMvc
class OrderServiceContractTest {

    @Autowired
    private MockMvc mockMvc;

    @Autowired
    private MonolithClient monolithClient;

    @ParameterizedTest
    @MethodSource("provideTestOrderIds")
    void shouldReturnSameResultAsMonolith(String orderId) throws Exception {
        // 从单体获取预期结果
        OrderDTO expected = monolithClient.getOrder(orderId);

        // 从新服务获取实际结果
        MvcResult result = mockMvc.perform(
                get("/api/v1/orders/{id}", orderId)
                    .accept(MediaType.APPLICATION_JSON))
            .andExpect(status().isOk())
            .andReturn();

        OrderDTO actual = objectMapper.readValue(
            result.getResponse().getContentAsString(),
            OrderDTO.class
        );

        // 逐字段比较
        assertThat(actual.getOrderId()).isEqualTo(expected.getOrderId());
        assertThat(actual.getStatus()).isEqualTo(expected.getStatus());
        assertThat(actual.getTotalAmount())
            .isCloseTo(expected.getTotalAmount(), within(0.01));
        assertThat(actual.getItems()).hasSize(expected.getItems().size());

        for (int i = 0; i < actual.getItems().size(); i++) {
            OrderItemDTO actualItem = actual.getItems().get(i);
            OrderItemDTO expectedItem = expected.getItems().get(i);
            assertThat(actualItem.getProductId())
                .isEqualTo(expectedItem.getProductId());
            assertThat(actualItem.getQuantity())
                .isEqualTo(expectedItem.getQuantity());
        }
    }

    static Stream<String> provideTestOrderIds() {
        return Stream.of(
            "ORD-001", "ORD-002", "ORD-003",
            "ORD-100", "ORD-999"
        );
    }
}

# packwerk 包定义文件 package.yml
# 位于 components/orders/package.yml
enforce_dependencies: true
enforce_privacy: true

dependencies:
  - components/products
  - components/customers
  - components/payments

# 明确声明公开 API
public_path: app/public/

# 不允许的依赖（违反则 CI 报错）
# components/orders 不应该直接依赖 components/shipping 的内部实现

# 边界检查的 CI 脚本
# scripts/check_boundaries.rb

require 'packwerk'

result = Packwerk::Cli.new.execute_command(['check'])

if result.include?('No violations detected')
  puts "边界检查通过"
  exit 0
else
  puts "发现边界违规："
  puts result
  exit 1
end

// Centrifuge 的核心设计：统一的 Destination 接口
// 所有 Destination 实现同一个接口，在单体内部注册

package centrifuge

// Destination 定义了所有数据目的地的统一接口
type Destination interface {
    // Name 返回目的地的唯一标识符
    Name() string

    // Send 将事件发送到目的地
    Send(ctx context.Context, event Event) error

    // BatchSend 批量发送事件
    BatchSend(ctx context.Context, events []Event) error

    // Validate 验证目的地配置
    Validate(config DestinationConfig) error
}

// Registry 管理所有已注册的 Destination
type Registry struct {
    mu           sync.RWMutex
    destinations map[string]Destination
}

func NewRegistry() *Registry {
    return &Registry{
        destinations: make(map[string]Destination),
    }
}

func (r *Registry) Register(d Destination) {
    r.mu.Lock()
    defer r.mu.Unlock()
    r.destinations[d.Name()] = d
}

func (r *Registry) Get(name string) (Destination, bool) {
    r.mu.RLock()
    defer r.mu.RUnlock()
    d, ok := r.destinations[name]
    return d, ok
}

// Worker 统一的事件处理 Worker
type Worker struct {
    registry    *Registry
    queue       Queue
    retrier     Retrier
    metrics     MetricsCollector
    rateLimiter RateLimiter
}

func (w *Worker) Process(ctx context.Context, msg Message) error {
    dest, ok := w.registry.Get(msg.DestinationName)
    if !ok {
        return fmt.Errorf("未知的目的地: %s", msg.DestinationName)
    }

    // 统一的速率限制
    if err := w.rateLimiter.Wait(ctx, msg.DestinationName); err != nil {
        return fmt.Errorf("速率限制: %w", err)
    }

    // 统一的重试逻辑
    err := w.retrier.Do(ctx, func() error {
        return dest.Send(ctx, msg.Event)
    })

    // 统一的指标上报
    w.metrics.RecordSend(msg.DestinationName, err)

    return err
}

// 使用异步事件替代同步调用
// 旧方式：同步调用（导致分布式单体）
public class OrderService {
    // 反模式：同步调用链
    public Order createOrder(CreateOrderRequest request) {
        // 同步调用库存服务
        inventoryClient.reserve(request.getItems());
        // 同步调用定价服务
        PriceResult price = pricingClient.calculate(request.getItems());
        // 同步调用用户服务
        UserProfile user = userClient.getProfile(request.getUserId());

        return orderRepository.save(
            new Order(request, price, user)
        );
    }
}

// 新方式：事件驱动（松耦合）
public class OrderService {
    private final EventPublisher eventPublisher;
    private final OrderRepository orderRepository;

    public Order createOrder(CreateOrderRequest request) {
        Order order = Order.createDraft(request);
        orderRepository.save(order);

        // 发布事件，异步处理后续流程
        eventPublisher.publish(new OrderCreatedEvent(
            order.getId(),
            order.getItems(),
            order.getCustomerId()
        ));

        return order;
    }

    // 监听其他服务发布的事件
    @EventListener
    public void onInventoryReserved(InventoryReservedEvent event) {
        Order order = orderRepository.findById(event.getOrderId());
        order.markInventoryReserved();
        orderRepository.save(order);
    }

    @EventListener
    public void onPaymentCompleted(PaymentCompletedEvent event) {
        Order order = orderRepository.findById(event.getOrderId());
        order.markPaid(event.getPaymentId());
        orderRepository.save(order);
    }
}

-- 旧系统：跨域的外键关系
-- orders 表引用 customers 表和 products 表
ALTER TABLE orders
    ADD CONSTRAINT fk_order_customer
    FOREIGN KEY (customer_id) REFERENCES customers(id);

ALTER TABLE order_items
    ADD CONSTRAINT fk_item_product
    FOREIGN KEY (product_id) REFERENCES products(id);

-- 迁移策略：用应用层校验替代数据库外键
-- 步骤一：移除外键约束
ALTER TABLE orders DROP FOREIGN KEY fk_order_customer;
ALTER TABLE order_items DROP FOREIGN KEY fk_item_product;

-- 步骤二：添加应用层验证（在服务代码中实现）
-- 步骤三：添加异步一致性检查任务

<!-- 反模式：所有服务共享同一个 common 库 -->
<!-- 修改 common 库意味着所有服务都需要重新构建和部署 -->
<dependency>
    <groupId>com.example</groupId>
    <artifactId>common-models</artifactId>
    <version>1.0.0</version> <!-- 所有服务使用同一版本 -->
</dependency>

<!-- 推荐：按职责拆分共享库，独立版本管理 -->
<dependency>
    <groupId>com.example</groupId>
    <artifactId>order-api-contracts</artifactId>
    <version>2.3.1</version> <!-- 订单服务的 API 契约 -->
</dependency>
<dependency>
    <groupId>com.example</groupId>
    <artifactId>common-observability</artifactId>
    <version>1.5.0</version> <!-- 可观测性工具包 -->
</dependency>

class DataMigrationStep:
    """可回退的数据迁移步骤"""

    def __init__(self, step_name: str, source_db, target_db):
        self.step_name = step_name
        self.source_db = source_db
        self.target_db = target_db
        self.checkpoint_table = f"migration_checkpoint_{step_name}"

    def forward(self, batch_size: int = 1000):
        """正向迁移：从源库迁移到目标库"""
        last_id = self._get_checkpoint()

        while True:
            rows = self.source_db.execute(
                f"SELECT * FROM orders WHERE id > %s "
                f"ORDER BY id LIMIT %s",
                (last_id, batch_size)
            )

            if not rows:
                break

            transformed = [self._transform(row) for row in rows]
            self.target_db.batch_insert("orders", transformed)

            last_id = rows[-1]["id"]
            self._save_checkpoint(last_id)

            print(f"[{self.step_name}] 已迁移至 id={last_id}")

    def rollback(self):
        """回退：清除目标库中已迁移的数据"""
        last_id = self._get_checkpoint()
        self.target_db.execute(
            "DELETE FROM orders WHERE id <= %s AND source = 'migration'",
            (last_id,)
        )
        self._reset_checkpoint()
        print(f"[{self.step_name}] 回退完成")

    def verify(self) -> bool:
        """校验：比较源库和目标库的数据一致性"""
        source_count = self.source_db.execute(
            "SELECT COUNT(*) as cnt FROM orders"
        )[0]["cnt"]
        target_count = self.target_db.execute(
            "SELECT COUNT(*) as cnt FROM orders WHERE source = 'migration'"
        )[0]["cnt"]

        source_checksum = self.source_db.execute(
            "SELECT MD5(GROUP_CONCAT(id, status, total_amount "
            "ORDER BY id)) as checksum FROM orders"
        )[0]["checksum"]
        target_checksum = self.target_db.execute(
            "SELECT MD5(GROUP_CONCAT(id, status, total_amount "
            "ORDER BY id)) as checksum FROM orders "
            "WHERE source = 'migration'"
        )[0]["checksum"]

        is_consistent = (
            source_count == target_count
            and source_checksum == target_checksum
        )
        print(
            f"[{self.step_name}] 校验结果: "
            f"源={source_count}, 目标={target_count}, "
            f"一致={is_consistent}"
        )
        return is_consistent

    def _transform(self, row: dict) -> dict:
        """数据转换逻辑"""
        return {
            "id": row["id"],
            "order_id": row["order_no"],
            "customer_id": str(row["user_id"]),
            "status": self._map_status(row["status_code"]),
            "total_amount": float(row["total"]),
            "source": "migration",
            "migrated_at": datetime.utcnow().isoformat()
        }

    def _map_status(self, code: int) -> str:
        status_map = {
            0: "DRAFT", 1: "PENDING", 2: "PAID",
            3: "SHIPPED", 4: "DELIVERED", 5: "CANCELLED"
        }
        return status_map.get(code, "UNKNOWN")

    def _get_checkpoint(self) -> int:
        result = self.target_db.execute(
            f"SELECT last_id FROM {self.checkpoint_table} LIMIT 1"
        )
        return result[0]["last_id"] if result else 0

    def _save_checkpoint(self, last_id: int):
        self.target_db.execute(
            f"INSERT INTO {self.checkpoint_table} (last_id, updated_at) "
            f"VALUES (%s, NOW()) "
            f"ON DUPLICATE KEY UPDATE last_id = %s, updated_at = NOW()",
            (last_id, last_id)
        )

    def _reset_checkpoint(self):
        self.target_db.execute(f"DELETE FROM {self.checkpoint_table}")

# ADR-007：订单服务数据库拆分策略

## 状态
已采纳（2026-03-15）

## 背景
订单模块已从单体代码中组件化，需要决定数据库拆分策略。

## 决策
采用三阶段迁移策略：
1. 阶段一（2026-04）：共享数据库，创建独立 Schema
2. 阶段二（2026-06）：引入 CDC，双写同步
3. 阶段三（2026-09）：切换到独立数据库

## 理由
- 三阶段策略风险可控
- CDC 保证迁移期数据一致性
- 每个阶段有明确的回退方案

## 后果
- 迁移期间增加 CDC 基础设施运维成本
- 需要开发双写校验工具
- 阶段二到阶段三期间存在数据延迟（预计小于 1 秒）

# 迁移进度追踪指标
migration_metrics:
  progress:
    - name: modules_migrated_ratio
      formula: "已迁移模块数 / 总模块数"
      target: "按季度递增"
      current: "4/12 = 33%"

    - name: traffic_migrated_ratio
      formula: "新服务处理的请求占比"
      target: "按月递增"
      current: "45%"

    - name: database_tables_migrated
      formula: "已迁移到独立库的表数量 / 总表数量"
      target: "按季度递增"
      current: "28/120 = 23%"

  quality:
    - name: dual_write_consistency
      formula: "双写数据一致的记录数 / 总记录数"
      target: "> 99.99%"
      current: "99.997%"

    - name: contract_test_coverage
      formula: "有契约测试的 API 端点数 / 总 API 端点数"
      target: "> 95%"
      current: "87%"

    - name: rollback_count
      formula: "本季度因迁移导致的回退次数"
      target: "< 3 次/季度"
      current: "1 次"

  efficiency:
    - name: new_service_deploy_time
      formula: "新服务从代码提交到生产部署的平均时间"
      target: "< 15 分钟"
      current: "12 分钟"

    - name: monolith_deploy_time
      formula: "单体应用的部署时间（应逐步缩短）"
      target: "逐季度缩短"
      current: "从 45 分钟缩短至 25 分钟"

@AnalyzeClasses(packages = "com.example.ecommerce")
public class ArchitectureBoundaryTest {

    @ArchTest
    static final ArchRule orderModuleShouldNotDependOnShipping =
        noClasses()
            .that().resideInAPackage("..order..")
            .should().dependOnClassesThat()
            .resideInAPackage("..shipping.internal..")
            .because("订单模块不应该依赖物流模块的内部实现，"
                    + "应通过 shipping.api 包中的接口通信");

    @ArchTest
    static final ArchRule servicesShouldNotAccessOtherServicesRepositories =
        noClasses()
            .that().resideInAPackage("..order..")
            .should().dependOnClassesThat()
            .resideInAPackage("..inventory.repository..")
            .because("服务不应该直接访问其他服务的数据访问层");

    @ArchTest
    static final ArchRule domainShouldNotDependOnInfrastructure =
        noClasses()
            .that().resideInAPackage("..domain..")
            .should().dependOnClassesThat()
            .resideInAnyPackage(
                "..infrastructure..",
                "..controller..",
                "org.springframework.."
            )
            .because("领域层不应依赖基础设施层（依赖反转原则）");

    @ArchTest
    static final ArchRule publicApiShouldBeInApiPackage =
        classes()
            .that().areAnnotatedWith(PublicApi.class)
            .should().resideInAPackage("..api..")
            .because("公开 API 类应统一放在 api 包中");
}