@Data
@Builder
public class KnowledgebaseSyncEvent {

    // 事件类型
    public enum EventType {
        DOCUMENT_ADDED,      // 新增文档
        DOCUMENT_UPDATED,    // 更新文档
        DOCUMENT_DELETED,    // 删除文档
        COLLECTION_REBUILT,  // 集合重建（Embedding 模型升级后）
        VECTOR_SYNC          // 直接同步向量（增量同步优化）
    }

    private String eventId;
    private EventType eventType;
    private String collectionName;
    private String documentId;
    private long version;           // 知识库版本号

    // 文档内容（原始）
    private String content;
    private Map<String, Object> metadata;

    // 向量数据（可选，用于直接向量同步）
    private float[] embedding;
    private String embeddingModel;   // 必须记录，方便接收方验证一致性
    private String embeddingModelVersion;

    // 发布信息
    private String sourceRegion;
    private long publishedAt;

    // 目标 Region（为空则广播到所有 Region）
    private List<String> targetRegions;
}

4.2 知识库写入服务（主 Region）

@Service
@Slf4j
public class GlobalKnowledgebaseService {

    @Autowired
    private VectorStoreClient localVectorStore;

    @Autowired
    private KafkaTemplate<String, KnowledgebaseSyncEvent> kafkaTemplate;

    @Autowired
    private EmbeddingService embeddingService;

    @Autowired
    private KnowledgebaseVersionService versionService;

    @Value("${app.region}")
    private String currentRegion;

    /**
     * 添加文档到知识库（主 Region 写入，同时触发跨 Region 同步）
     */
    @Transactional
    public AddDocumentResult addDocument(String collectionName, KnowledgeDocument document) {
        long newVersion = versionService.incrementVersion(collectionName);

        // 1. 计算向量
        float[] embedding = embeddingService.embed(document.getContent());
        String embeddingModel = embeddingService.getCurrentModel();
        String embeddingModelVersion = embeddingService.getCurrentModelVersion();

        // 2. 写入本地向量数据库
        String vectorId = localVectorStore.upsert(
            collectionName,
            document.getId(),
            embedding,
            document.getMetadata()
        );

        // 3. 发布同步事件到 Kafka
        KnowledgebaseSyncEvent event = KnowledgebaseSyncEvent.builder()
                .eventId(UUID.randomUUID().toString())
                .eventType(KnowledgebaseSyncEvent.EventType.VECTOR_SYNC)
                .collectionName(collectionName)
                .documentId(document.getId())
                .version(newVersion)
                .content(document.getContent())     // 原始文档也一起发（备用）
                .metadata(document.getMetadata())
                .embedding(embedding)               // 向量直接传播，节省从 Region 的 Embedding 成本
                .embeddingModel(embeddingModel)
                .embeddingModelVersion(embeddingModelVersion)
                .sourceRegion(currentRegion)
                .publishedAt(System.currentTimeMillis())
                .build();

        // 发送到各 Region 专属的 Topic
        String topicName = "kb-sync-global";
        kafkaTemplate.send(topicName, collectionName, event)
                .addCallback(
                    result -> log.debug("Sync event published: eventId={}, version={}",
                            event.getEventId(), event.getVersion()),
                    ex -> log.error("Failed to publish sync event: {}", ex.getMessage(), ex)
                );

        log.info("Document added and sync event published: docId={}, collection={}, version={}",
                document.getId(), collectionName, newVersion);

        return AddDocumentResult.builder()
                .documentId(document.getId())
                .vectorId(vectorId)
                .version(newVersion)
                .build();
    }
}

4.3 从 Region 同步消费者

@Component
@Slf4j
public class KnowledgebaseSyncConsumer {

    @Autowired
    private VectorStoreClient localVectorStore;

    @Autowired
    private EmbeddingService embeddingService;

    @Autowired
    private KnowledgebaseVersionService versionService;

    @Autowired
    private MeterRegistry meterRegistry;

    @Value("${app.region}")
    private String currentRegion;

    @KafkaListener(
        topics = "kb-sync-global",
        groupId = "${app.region}-kb-sync-consumer",
        concurrency = "3"
    )
    public void handleSyncEvent(KnowledgebaseSyncEvent event, Acknowledgment ack) {
        // 主 Region 不处理自己发出的事件
        if (currentRegion.equals(event.getSourceRegion())) {
            ack.acknowledge();
            return;
        }

        log.debug("Received sync event: type={}, doc={}, from={}",
                event.getEventType(), event.getDocumentId(), event.getSourceRegion());

        try {
            processSyncEvent(event);
            ack.acknowledge();

            meterRegistry.counter("kb.sync.success",
                    "region", currentRegion,
                    "event_type", event.getEventType().name()
            ).increment();

        } catch (Exception e) {
            log.error("Failed to process sync event {}: {}", event.getEventId(), e.getMessage(), e);

            meterRegistry.counter("kb.sync.failed",
                    "region", currentRegion,
                    "event_type", event.getEventType().name()
            ).increment();

            // 不 ack，Kafka 会重试
            throw new RuntimeException("Sync event processing failed", e);
        }
    }

    private void processSyncEvent(KnowledgebaseSyncEvent event) {
        switch (event.getEventType()) {
            case VECTOR_SYNC, DOCUMENT_ADDED -> handleVectorSync(event);
            case DOCUMENT_UPDATED -> handleVectorSync(event);   // 更新等同于覆盖
            case DOCUMENT_DELETED -> handleDocumentDelete(event);
            case COLLECTION_REBUILT -> handleCollectionRebuild(event);
            default -> log.warn("Unknown event type: {}", event.getEventType());
        }
    }

    private void handleVectorSync(KnowledgebaseSyncEvent event) {
        // 检查 Embedding 模型版本是否一致
        if (!embeddingService.getCurrentModel().equals(event.getEmbeddingModel()) ||
            !embeddingService.getCurrentModelVersion().equals(event.getEmbeddingModelVersion())) {
            log.warn("Embedding model mismatch! Local: {}/{}, Event: {}/{}. " +
                    "Falling back to re-embedding.",
                    embeddingService.getCurrentModel(), embeddingService.getCurrentModelVersion(),
                    event.getEmbeddingModel(), event.getEmbeddingModelVersion());

            // 模型不一致时，本地重新 Embedding（保证向量空间一致）
            float[] localEmbedding = embeddingService.embed(event.getContent());
            event = event.toBuilder().embedding(localEmbedding).build();

            // 记录不一致告警
            meterRegistry.counter("kb.sync.model_mismatch",
                    "region", currentRegion).increment();
        }

        // 写入本地向量数据库
        localVectorStore.upsert(
            event.getCollectionName(),
            event.getDocumentId(),
            event.getEmbedding(),
            event.getMetadata()
        );

        // 更新本地版本记录
        versionService.updateVersion(event.getCollectionName(), event.getVersion());

        log.debug("Vector synced: docId={}, collection={}, version={}",
                event.getDocumentId(), event.getCollectionName(), event.getVersion());
    }

    private void handleDocumentDelete(KnowledgebaseSyncEvent event) {
        localVectorStore.delete(event.getCollectionName(), event.getDocumentId());
        versionService.updateVersion(event.getCollectionName(), event.getVersion());
        log.info("Document deleted from vector store: {}", event.getDocumentId());
    }

    private void handleCollectionRebuild(KnowledgebaseSyncEvent event) {
        // 集合重建事件，触发本地全量同步
        log.warn("Collection rebuild event received for {}, starting full sync...",
                event.getCollectionName());
        // 实际实现需要触发一个异步任务从 OSS 拉取完整文档集合重建
        // 这里省略具体实现，实际项目中这是一个独立的任务
    }
}

4.4 同步延迟监控

@Component
@Slf4j
public class SyncLagMonitor {

    @Autowired
    private KnowledgebaseVersionService versionService;

    @Autowired
    private GlobalVersionService globalVersionService;

    @Autowired
    private MeterRegistry meterRegistry;

    @Value("${app.region}")
    private String currentRegion;

    // 每分钟检查一次同步延迟
    @Scheduled(fixedDelay = 60000)
    public void checkSyncLag() {
        List<String> collections = List.of("customer_service_kb", "product_kb", "faq_kb");

        for (String collection : collections) {
            long globalVersion = globalVersionService.getGlobalVersion(collection);
            long localVersion = versionService.getLocalVersion(collection);
            long lag = globalVersion - localVersion;

            meterRegistry.gauge("kb.sync.lag",
                    Tags.of("region", currentRegion, "collection", collection),
                    lag);

            if (lag > 100) {
                log.warn("Sync lag for collection {} in region {}: {} versions behind",
                        collection, currentRegion, lag);
            }
        }
    }
}

五、Embedding 模型升级的全球协调

这是多 Region 部署里最复杂的操作之一。需要一个协调流程：

@Service
@Slf4j
public class GlobalEmbeddingUpgradeCoordinator {

    @Autowired
    private List<RegionClient> regionClients;

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    /**
     * 全球 Embedding 模型升级流程
     * 必须严格按顺序执行，不能并发
     */
    @Async
    public void upgradeEmbeddingModel(String newModel, String newModelVersion) {
        log.info("Starting global Embedding model upgrade: {} -> {}", 
                embeddingService.getCurrentModel(), newModel);

        // 第一步：暂停所有 Region 的知识库写入
        log.info("Step 1: Pausing knowledge base writes in all regions");
        for (RegionClient region : regionClients) {
            region.pauseKnowledgebaseWrites();
        }

        // 等待所有在途的同步事件处理完
        log.info("Step 2: Waiting for in-flight sync events to complete");
        waitForSyncQueueDrained();

        // 第三步：更新所有 Region 的 Embedding 模型版本配置
        log.info("Step 3: Updating Embedding model config in all regions");
        for (RegionClient region : regionClients) {
            region.updateEmbeddingModel(newModel, newModelVersion);
        }

        // 第四步：在主 Region 全量重建向量
        log.info("Step 4: Rebuilding vectors in primary region");
        rebuildAllVectors(newModel);

        // 第五步：全量同步到所有从 Region
        log.info("Step 5: Full sync to all secondary regions");
        triggerGlobalFullSync();

        // 第六步：恢复所有 Region 的写入
        log.info("Step 6: Resuming knowledge base writes");
        for (RegionClient region : regionClients) {
            region.resumeKnowledgebaseWrites();
        }

        log.info("Global Embedding model upgrade completed");
    }

    private void waitForSyncQueueDrained() {
        // 等待 Kafka topic 的 consumer lag 为 0
        long maxWaitMs = 300000; // 最多等 5 分钟
        long startTime = System.currentTimeMillis();

        while (System.currentTimeMillis() - startTime < maxWaitMs) {
            long totalLag = kafkaAdminClient.getConsumerGroupLag("kb-sync-global");
            if (totalLag == 0) {
                log.info("Sync queue drained");
                return;
            }
            log.info("Waiting for sync queue to drain, current lag: {}", totalLag);
            sleep(5000);
        }

        throw new RuntimeException("Sync queue did not drain within timeout");
    }

    private void sleep(long ms) {
        try { Thread.sleep(ms); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
    }
}

六、Region 路由策略

最后，用户请求如何路由到正确的 Region？

最简单的方式是 DNS GeoDNS：根据用户 IP 解析到最近 Region 的 IP。

但有几个特殊情况需要处理：

@Component
public class RegionRouter {

    /**
     * 确定请求应该路由到哪个 Region
     * 返回 null 表示使用默认区域（通常是地理最近的）
     */
    public String determineTargetRegion(HttpServletRequest request, String userId) {
        // 规则1：如果用户所在组织有固定的 Region 绑定，使用组织绑定
        String orgRegion = userOrgService.getOrgRegion(userId);
        if (orgRegion != null) {
            return orgRegion;
        }

        // 规则2：如果请求头里指定了 Region（通常由前端根据 GeoDNS 结果传入）
        String requestedRegion = request.getHeader("X-Preferred-Region");
        if (requestedRegion != null && isValidRegion(requestedRegion)) {
            return requestedRegion;
        }

        // 规则3：检查目标区域的知识库同步状态
        // 如果目标区域同步严重滞后，路由到主区域
        String geoRegion = geoService.getRegionByIp(request.getRemoteAddr());
        if (geoRegion != null) {
            long lag = syncLagMonitor.getSyncLag(geoRegion, "customer_service_kb");
            if (lag > 500) {
                log.warn("Region {} has high sync lag ({}), falling back to primary region",
                        geoRegion, lag);
                return primaryRegion;
            }
            return geoRegion;
        }

        return null; // 使用默认
    }

    private boolean isValidRegion(String region) {
        return Set.of("ap-southeast-1", "us-east-1", "eu-west-1").contains(region);
    }
}

总结

AI 应用的多 Region 部署比普通应用难，核心原因是：

Embedding 模型版本必须全球一致，向量才能互通
知识库同步延迟会导致不同 Region 返回不同结果
Embedding 模型升级需要全球协调，不能像普通配置那样各自更新

三种部署模式的选择：

主写从读：一致性强，写延迟高
各 Region 独立：性能最好，一致性最弱
混合模式：最灵活，最复杂

基于消息队列（Kafka）的向量同步是实现主写从读模式的核心技术，关键点是同步事件里必须携带 Embedding 模型版本信息，从 Region 收到后验证版本一致性。

不要低估多 Region 的复杂度。如果业务还没到需要多 Region 的规模，先做好单 Region 的高可用，多 Region 是在这个基础上的额外投入。