16-项目经验

┌─────────────────────────────────────────────────┐
│                  接入层                          │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐      │
│  │  Nginx   │  │API Gateway│  │  负载均衡  │      │
│  └──────────┘  └──────────┘  └──────────┘      │
└─────────────────────┬───────────────────────────┘
                      ↓
┌─────────────────────────────────────────────────┐
│                  业务层                          │
│  ┌─────────────┐  ┌──────────────┐              │
│  │  会话服务    │  │  权限服务     │              │
│  └─────────────┘  └──────────────┘              │
│  ┌─────────────┐  ┌──────────────┐              │
│  │  Agent编排   │  │  工具服务     │              │
│  └─────────────┘  └──────────────┘              │
│  ┌─────────────┐  ┌──────────────┐              │
│  │  知识库服务   │  │  消息通知     │              │
│  └─────────────┘  └──────────────┘              │
└─────────────────────┬───────────────────────────┘
                      ↓
┌─────────────────────────────────────────────────┐
│                  AI 层                           │
│  ┌─────────────┐  ┌──────────────┐              │
│  │ 大模型服务   │  │ RAG 检索服务  │              │
│  └─────────────┘  └──────────────┘              │
│  ┌─────────────┐  ┌──────────────┐              │
│  │ Prompt管理  │  │ 可观测平台    │              │
│  └─────────────┘  └──────────────┘              │
└─────────────────────┬───────────────────────────┘
                      ↓
┌─────────────────────────────────────────────────┐
│                  数据层                          │
│  ┌─────────────┐  ┌──────────────┐              │
│  │ MySQL 主从  │  │ Redis 集群    │              │
│  └─────────────┘  └──────────────┘              │
│  ┌─────────────┐  ┌──────────────┐              │
│  │  ElasticSearch│  │ Milvus 向量库 │              │
│  └─────────────┘  └──────────────┘              │
└─────────────────────────────────────────────────┘

S（Situation）- 背景
├── 项目初期 RAG 效果差
├── 用户反馈回答不准确
├── 数据：Top 5 召回率 72%
└── 影响：客服效率低，用户投诉多

T（Task）- 任务
├── 目标：将 Top 5 召回率提升到 90%+
├── 约束：不改业务逻辑，纯优化
└── 时间：2 周

A（Action）- 行动
├── 1. 构建评测数据集（2000 条标注）
├── 2. Chunk 语义切分（语义边界 + 50% 重叠）
├── 3. 混合检索融合（BM25 + 向量 + RRF）
├── 4. Cross-Encoder 重排序
└── 5. Query 动态扩展

R（Result）- 结果
├── Top 5 召回率：72% → 93%
├── 用户满意度：68% → 89%
├── 客服平均处理时间：8 分钟 → 3 分钟
└── 获得季度项目之星

// 混合检索核心实现
public class HybridRetrievalEngine {

    private final BM25Search bm25;
    private final VectorSearch vectorSearch;
    private final CrossEncoderReranker reranker;

    public RetrievalResult retrieve(String query) {
        // Step 1: Query 扩展
        String expandedQuery = queryExpander.expand(query);

        // Step 2: 双路召回
        List<ScoredDoc> bm25Results = bm25.search(expandedQuery, 50);
        List<ScoredDoc> vectorResults = vectorSearch.search(expandedQuery, 50);

        // Step 3: RRF 融合
        List<ScoredDoc> fused = rrfFusion(bm25Results, vectorResults);

        // Step 4: Cross-Encoder 重排序
        List<ScoredDoc> reranked = reranker.rerank(expandedQuery, fused, 20);

        return new RetrievalResult(reranked);
    }

    // Reciprocal Rank Fusion
    private List<ScoredDoc> rrfFusion(List<ScoredDoc> list1,
            List<ScoredDoc> list2) {
        Map<String, Double> rrfScores = new HashMap<>();
        int k = 60;

        for (int i = 0; i < list1.size(); i++) {
            rrfScores.merge(list1.get(i).getId(),
                1.0 / (k + i + 1), Double::sum);
        }
        for (int i = 0; i < list2.size(); i++) {
            rrfScores.merge(list2.get(i).getId(),
                1.0 / (k + i + 1), Double::sum);
        }

        return rrfScores.entrySet().stream()
            .sorted(Map.Entry.<String, Double>comparingByValue().reversed())
            .map(e -> new ScoredDoc(e.getKey(), e.getValue()))
            .collect(Collectors.toList());
    }
}

故障现象：
├── 高峰期整点（每小时）请求超时
├── 数据库连接数达 200 上限
└── 应用日志：Too many connections

排查步骤：
Step 1: 确认时间规律
├── 观察到每小时整点触发
├── 整点有定时任务 + 业务高峰
└── 初步怀疑：定时任务或高并发导致

Step 2: Arthas 现场诊断
├── thread -n 3: 观察到大量线程卡在 getConnection()
├── trace DataSource.getConnection: 连接获取耗时超长
└── 定位到 Agent 工具模块的数据库操作

Step 3: 代码审查
├── Agent 工具用 @Transactional 注解
├── 调用链经过异步线程（@Async）
├── @Transactional 默认不传播到异步线程
└── 事务未提交 → 连接未释放 → 连接池耗尽

Step 4: 验证假设
├── 手动触发 Agent 工具调用
├── 监控连接池变化
└── 确认：每次调用增加 3-5 个未释放连接

// 方案一：使用编程式事务（确保事务在当前线程提交）
@Service
public class ToolDataService {

    private final TransactionTemplate transactionTemplate;

    public void executeInTransaction(Runnable operation) {
        transactionTemplate.execute(status -> {
            operation.run();
            return null;
        });
    }
}

// 方案二：架构调整 —— 将数据库操作剥离到独立服务
@Service
public class OrderToolService {

    private final OrderRepository orderRepository;
    private final ToolDataService toolDataService;

    public ToolResult queryOrder(String orderId) {
        return toolDataService.executeInTransaction(() -> {
            Order order = orderRepository.findById(orderId)
                .orElseThrow(() -> new BusinessException("订单不存在"));
            return ToolResult.success(order);
        });
    }
}

// 方案三：连接池监控告警
@Configuration
public class ConnectionPoolMonitor {

    @Scheduled(fixedRate = 5000) // 每 5 秒检查
    public void monitor() {
        HikariPoolMXBean pool = getHikariPoolMXBean();
        int active = pool.getActiveConnections();
        int idle = pool.getIdleConnections();
        int waiting = pool.getThreadsAwaitingConnection();

        if (waiting > 0 || active > 150) {
            alertService.alert("数据库连接池告警",
                String.format("活跃:%d 空闲:%d 等待:%d", active, idle, waiting));
        }
    }
}

需求分析
    ↓
┌─────────────────────┐
│ 1. 明确选型标准      │
│    - 核心需求        │
│    - 团队能力        │
│    - 运维成本        │
│    - 社区生态        │
└─────────┬───────────┘
          ↓
┌─────────────────────┐
│ 2. 候选方案调研      │
│    - 技术文档阅读    │
│    - 社区口碑调研    │
│    - 同类项目参考    │
└─────────┬───────────┘
          ↓
┌─────────────────────┐
│ 3. POC 验证          │
│    - 搭建 Demo       │
│    - 性能测试        │
│    - 集成测试        │
│    - 团队技术评估    │
└─────────┬───────────┘
          ↓
┌─────────────────────┐
│ 4. 决策与落地        │
│    - 技术评审会      │
│    - 团队培训        │
│    - 灰度使用        │
│    - 全量推广        │
└─────────────────────┘

@Service
public class AgentOrchestrationService {

    private final ChatModel chatModel;
    private final ToolRegistry toolRegistry;
    private final RAGRetrievalService ragService;
    private final ContextManager contextManager;

    public AgentResponse processUserMessage(Session session, String userMessage) {
        // Step 1: 构建上下文
        List<Message> messages = contextManager.buildContext(session);
        messages.add(new UserMessage(userMessage));

        // Step 2: RAG 知识检索
        List<Document> knowledge = ragService.retrieve(userMessage);
        if (!knowledge.isEmpty()) {
            messages.add(new SystemMessage(
                "参考以下知识回答：" + formatDocuments(knowledge)
            ));
        }

        // Step 3: 调用大模型（ReAct 循环）
        ChatResponse response;
        int maxIterations = 5;

        for (int i = 0; i < maxIterations; i++) {
            response = chatModel.chat(messages);

            if (response.isToolCall()) {
                // 工具调用
                ToolResult toolResult = executeToolCall(response.getToolCall());
                messages.add(new AssistantMessage(response.getText()));
                messages.add(new ToolMessage(toolResult.toString()));
            } else {
                // 最终回复
                break;
            }
        }

        // Step 4: 保存上下文
        contextManager.saveContext(session, messages);

        return AgentResponse.from(response);
    }
}

@Service
public class RAGRetrievalService {

    private final BM25Search bm25Search;
    private final VectorSearch vectorSearch;
    private final CrossEncoderReranker reranker;

    public List<Document> retrieve(String query) {
        // 双路召回
        List<ScoredDoc> bm25Results = bm25Search.search(query, 50);
        List<ScoredDoc> vectorResults = vectorSearch.search(query, 50);

        // RRF 融合
        List<ScoredDoc> fused = rrfFusion(bm25Results, vectorResults);

        // Cross-Encoder 重排序
        return reranker.rerank(query, fused, 5);
    }

    public void indexDocument(String docId, String content) {
        ChunkService chunkService = new ChunkService();
        List<Chunk> chunks = chunkService.semanticChunk(content);

        for (Chunk chunk : chunks) {
            // 同时写入 ES 和 Milvus
            bm25Search.index(docId, chunk.getText());
            vectorSearch.index(docId, chunk.getText());
        }
    }
}

// 热点 Key 散列
public class HotKeyDispatcher {

    // 将热点 Key 散列到不同 Key，避免单 Key 过热
    public String disperseKey(String originalKey) {
        int slot = Math.abs(originalKey.hashCode()) % 16;
        return originalKey + ":slot" + slot;
    }

    // 二级缓存：本地缓存 + Redis
    @Cacheable(cacheNames = "local", key = "#root.args[0]")
    public Product getProduct(String productId) {
        return productRepository.findById(productId);
    }
}

-- 添加复合索引优化查询
ALTER TABLE orders
ADD INDEX idx_user_status_create (user_id, status, create_time);

-- 使用读写分离
@DataSource("slave")  // 读操作走从库
public List<Order> queryByUserId(Long userId) {
    return orderRepository.findByUserId(userId);
}

┌─────────────────────────────────────────────────┐
│              架构层优化                          │
│  CDN 静态化 | MQ 削峰 | 服务降级                 │
└─────────────────────┬───────────────────────────┘
                      ↓
┌─────────────────────────────────────────────────┐
│              应用层优化                          │
│  对象池复用 | 异步化 | 本地缓存 | 代码优化       │
└─────────────────────┬───────────────────────────┘
                      ↓
┌─────────────────────────────────────────────────┐
│              缓存层优化                          │
│  Redis Cluster | 多级缓存 | 热点 Key | 穿透防护  │
└─────────────────────┬───────────────────────────┘
                      ↓
┌─────────────────────────────────────────────────┐
│              数据库层优化                        │
│  分库分表 | 索引优化 | SQL 调优 | 读写分离       │
└─────────────────────────────────────────────────┘

// ShardingSphere 分库分表配置
@Configuration
public class ShardingConfig {

    @Bean
    public ShardingRuleConfiguration shardingRule() {
        ShardingRuleConfiguration rule = new ShardingRuleConfiguration();
        rule.setTableShardingStrategyConfigs(Arrays.asList(
            new TableShardingStrategyConfiguration(
                "t_order",
                "user_id",
                "mod"
            )
        ));
        // 按 user_id 分 10 库 100 表
        rule.setDatabaseShardingStrategyConfigs(...);
        return rule;
    }
}

@Service
public class MultiLevelCacheService {

    private final CacheManager localCacheManager;  // Caffeine 本地缓存
    private final RedisTemplate redisTemplate;      // Redis 分布式缓存

    public Object getData(String key) {
        // L1: 本地缓存（Caffeine，10 分钟过期）
        Cache localCache = localCacheManager.getCache("local");
        Object value = localCache.get(key);
        if (value != null) return value;

        // L2: Redis 缓存（1 小时过期）
        value = redisTemplate.opsForValue().get(key);
        if (value != null) {
            localCache.put(key, value);  // 回填本地缓存
            return value;
        }

        // L3: 数据库查询
        value = queryFromDatabase(key);

        // 写入两级缓存
        redisTemplate.opsForValue().set(key, value, 1, TimeUnit.HOURS);
        localCache.put(key, value);

        return value;
    }
}

// 核心接口异步化
@Service
public class OrderService {

    @Async("orderExecutor")
    public CompletableFuture<OrderResult> createOrderAsync(OrderDTO dto) {
        // 同步逻辑改为异步执行
        Order order = orderRepository.save(convertToEntity(dto));

        // 异步发送通知
        notificationService.sendOrderCreatedNotification(order);

        return CompletableFuture.completedFuture(
            new OrderResult(order.getId(), "SUCCESS")
        );
    }

    // 批量并行处理
    public List<CompletableFuture<OrderResult>> batchCreateOrders(
            List<OrderDTO> dtos) {
        return dtos.stream()
            .map(this::createOrderAsync)
            .collect(Collectors.toList());
    }
}

// 幂等注解
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface Idempotent {
    String keyPrefix() default "idempotent:";
    int expireSeconds() default 300;
}

// 幂等切面
@Aspect
@Component
public class IdempotentAspect {

    private final RedisTemplate redisTemplate;

    @Around("@annotation(idempotent)")
    public Object around(ProceedingJoinPoint pjp, Idempotent idempotent)
            throws Throwable {
        String token = extractToken(pjp);  // 从请求头获取
        String key = idempotent.keyPrefix() + token;

        // 检查是否已处理
        Boolean exists = redisTemplate.hasKey(key);
        if (Boolean.TRUE.equals(exists)) {
            // 已处理，直接返回缓存的结果
            return redisTemplate.opsForValue().get(key + ":result");
        }

        // 加锁并执行
        Boolean locked = redisTemplate.opsForValue()
            .setIfAbsent(key, "processing", 10, TimeUnit.SECONDS);
        if (Boolean.FALSE.equals(locked)) {
            throw new BusinessException("请求正在处理中，请稍后重试");
        }

        try {
            Object result = pjp.proceed();
            // 缓存结果
            redisTemplate.opsForValue().set(key + ":result", result,
                idempotent.expireSeconds(), TimeUnit.SECONDS);
            return result;
        } finally {
            redisTemplate.delete(key);  // 释放锁
        }
    }
}

-- 创建型接口的唯一索引
ALTER TABLE orders
ADD UNIQUE KEY uk_order_user (order_no, user_id);

-- 状态机校验
@Service
public class OrderService {

    @Idempotent(keyPrefix = "order:create:")
    public OrderResult createOrder(OrderCreateDTO dto) {
        // 数据库唯一索引防止重复创建
        try {
            Order order = orderRepository.save(convertToEntity(dto));
            return OrderResult.success(order.getId());
        } catch (DuplicateKeyException e) {
            // 唯一索引冲突，返回已存在的订单
            Order existing = orderRepository.findByOrderNo(dto.getOrderNo());
            return OrderResult.success(existing.getId());
        }
    }
}

// 支付接口的状态机幂等
@Service
public class PaymentService {

    private final Map<OrderStatus, Set<OrderStatus>> transitions = Map.of(
        OrderStatus.PENDING_PAYMENT, Set.of(OrderStatus.PAID),
        OrderStatus.PAID, Set.of(),  // 已支付，不再流转
        OrderStatus.CANCELLED, Set.of()  // 已取消
    );

    @Idempotent(keyPrefix = "payment:")
    public PaymentResult payOrder(String orderId, PaymentDTO dto) {
        Order order = orderRepository.findById(orderId);

        // 状态机校验
        if (!transitions.getOrDefault(order.getStatus(), Set.of())
                .contains(OrderStatus.PAID)) {
            // 状态不允许变更，返回当前状态
            return PaymentResult.currentStatus(order.getStatus());
        }

        // 执行支付...
        order.setStatus(OrderStatus.PAID);
        orderRepository.save(order);

        return PaymentResult.success();
    }
}

客服 A 看到的对话：
  用户：我的订单什么时候能发货？
  Agent：您的订单【ORD12345】预计明天发货
  用户：能帮我查一下物流吗？
  Agent：（错误回复）您的订单【ORD67890】已签收  ← 这是客服 B 的订单！

Step 1: 怀疑大模型问题
├── 检查是否大模型返回错误
├── 对比正确/错误请求的大模型输入
└── 发现：大模型输入中包含了错误的上下文 → 不是大模型的问题

Step 2: 检查上下文存储
├── 查看 Redis 中 sessionId 对应的数据
├── 发现：有时两个不同的会话指向同一个 Redis Key
└── 定位：sessionId 生成逻辑有问题

Step 3: 分析 sessionId 生成
├── 原代码：
└── String sessionId = userId + "_" + System.currentTimeMillis();

Step 4: 验证假设
├── 高并发下，同一毫秒内多个请求
├── System.currentTimeMillis() 返回相同值
├── 导致 sessionId 重复 → 上下文串扰
└── 确认根因

// 修复前：
public String generateSessionId(Long userId) {
    return userId + "_" + System.currentTimeMillis();
    // 高并发下毫秒级重复！
}

// 修复后：
public String generateSessionId(Long userId) {
    return userId + "_" + UUID.randomUUID().toString().replace("-", "");
    // UUID 保证全局唯一
}

// 额外保障：Redis SETNX 确保唯一性
public void saveSession(String sessionId, SessionData data) {
    Boolean success = redisTemplate.opsForValue()
        .setIfAbsent(sessionId, serialize(data), 24, TimeUnit.HOURS);
    if (Boolean.FALSE.equals(success)) {
        log.warn("Session ID冲突: {}", sessionId);
        // 重新生成
        sessionId = generateSessionId(userId);
    }
}

// 消费者去重
@RocketMQMessageListener(topic = "order_topic",
                         consumerGroup = "order_group")
public class OrderConsumer implements MessageListener {

    private final RedisTemplate redisTemplate;
    private final OrderRepository orderRepository;

    @Override
    public void onMessage(Message msg) {
        String msgId = msg.getMsgId();
        String dedupKey = "order:consumed:" + msgId;

        // Redis 去重（SETNX）
        Boolean isFirst = redisTemplate.opsForValue()
            .setIfAbsent(dedupKey, "1", 24, TimeUnit.HOURS);

        if (Boolean.FALSE.equals(isFirst)) {
            log.warn("消息已消费，跳过: {}", msgId);
            return;
        }

        // 数据库唯一索引二次保障
        try {
            processMessage(msg);
        } catch (DuplicateKeyException e) {
            log.warn("重复消息，跳过: {}", msgId);
        }
    }
}

// 积压处理方案
@Service
public class MessageBacklogHandler {

    // 1. 临时扩容消费者（通过 API 动态调整实例数）
    public void scaleUpConsumers(int targetInstances) {
        // 调用 K8s API 增加消费者 Pod 数量
        k8sClient.scaleDeployment("order-consumer", targetInstances);
    }

    // 2. 非核心消息降级
    public void processWithPriority(List<Message> messages) {
        List<Message> critical = messages.stream()
            .filter(m -> m.getPriority() == Priority.HIGH)
            .collect(Collectors.toList());

        List<Message> normal = messages.stream()
            .filter(m -> m.getPriority() == Priority.NORMAL)
            .collect(Collectors.toList());

        // 优先处理高优先级消息
        critical.forEach(this::processMessage);

        // 普通消息批量处理
        processBatch(normal);
    }

    // 3. 批量处理减少开销
    private void processBatch(List<Message> messages) {
        List<String> orderIds = messages.stream()
            .map(Message::getOrderId)
            .collect(Collectors.toList());

        // 批量数据库操作
        orderRepository.batchUpdateStatus(orderIds, Status.PROCESSING);
    }
}

// 生产者：按订单 ID 选择队列
public void sendOrderMessage(OrderEvent event) {
    rocketMQTemplate.asyncSend(
        "order_topic",
        MessageBuilder.withPayload(event)
            .setHeader("orderId", event.getOrderId())
            .build(),
        new SendCallback() { ... },
        // 关键：按 orderId 选择 MessageQueue
        new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> mqs,
                    Message msg, Object arg) {
                String orderId = (String) arg;
                int index = Math.abs(orderId.hashCode()) % mqs.size();
                return mqs.get(index);
            }
        },
        event.getOrderId()
    );
}

// 消费者：单线程消费保证顺序
@RocketMQMessageListener(
    topic = "order_topic",
    consumerGroup = "order_orderly_group",
    // 消费模式：顺序消费
    messageModel = MessageModel.CLUSTERING
)
public class OrderOrderlyConsumer implements MessageListenerOrderly {

    @Override
    public boolean consumeOrderly(MessageExt msg,
            ConsumeOrderlyContext context) {
        // 单线程处理，天然保证同一队列内消息顺序
        processMessage(msg);
        return true;
    }
}

// RBAC 数据模型
@Entity
public class SysUser {
    private Long id;
    private String username;
    private String password;
}

@Entity
public class SysRole {
    private Long id;
    private String name;
    private String code;
}

@Entity
public class SysPermission {
    private Long id;
    private String code;  // 如 "order:query"
    private String type;  // MENU / BUTTON / API
}

// Spring Security 配置
@Configuration
@EnableWebSecurity
@EnableGlobalMethodSecurity(prePostEnabled = true)
public class SecurityConfig {

    @Bean
    public SecurityFilterChain filterChain(HttpSecurity http) {
        http
            .addFilterBefore(jwtAuthFilter(), UsernamePasswordAuthenticationFilter.class)
            .authorizeRequests()
            .antMatchers("/api/admin/**").hasRole("ADMIN")
            .antMatchers("/api/order/**").hasAuthority("order:query")
            .anyRequest().authenticated();
    }
}

// 数据权限注解
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface DataScope {
    String deptAlias() default "d";   // 部门表别名
    String userAlias() default "u";   // 用户表别名
}

// 数据权限切面
@Aspect
@Component
public class DataScopeAspect {

    @Before("@annotation(dataScope)")
    public void doBefore(JoinPoint point, DataScope dataScope) {
        LoginUser user = SecurityUtils.getCurrentUser();

        // 构建数据权限 SQL 条件
        StringBuilder sql = new StringBuilder(" WHERE 1=1 ");

        if (user.isAdmin()) {
            sql.append(" ");  // 管理员看全部
        } else if (user.isDeptManager()) {
            // 部门经理看本部门
            sql.append(" AND dept_id = ").append(user.getDeptId());
        } else {
            // 普通员工看自己
            sql.append(" AND create_by = ").append(user.getUserId());
        }

        // 将条件放入 ThreadLocal，供后续 SQL 使用
        DataScopeHolder.set(sql.toString());
    }
}

// Service 使用示例
@Service
public class OrderService {

    @DataScope(deptAlias = "o", userAlias = "u")
    public List<Order> queryOrders(OrderQueryDTO dto) {
        MybatisPlusWrapper<Order> wrapper = new MybatisPlusWrapper<>();
        // 自动拼接数据权限条件
        String dataScopeSql = DataScopeHolder.get();
        wrapper.apply(dataScopeSql);
        return orderMapper.selectList(wrapper);
    }
}

请求到达
    ↓
┌─────────────────────────────┐
│  Layer 1: 认证（Authentication）│
│  JWT Token 验证              │
│  获取用户身份信息             │
└─────────────┬───────────────┘
              ↓
┌─────────────────────────────┐
│  Layer 2: 功能权限（RBAC）    │
│  @PreAuthorize 注解校验       │
│  是否有接口访问权限           │
│  → Controller 层拦截          │
└─────────────┬───────────────┘
              ↓
┌─────────────────────────────┐
│  Layer 3: 数据权限（DataScope）│
│  @DataScope 注解 + AOP        │
│  拼接数据过滤条件             │
│  → Service/Mapper 层过滤     │
└─────────────────────────────┘

// 错误码体系
public enum ErrorCode {
    // 订单相关（1000-1999）
    ORDER_NOT_FOUND(1001, "订单不存在"),
    ORDER_STATUS_INVALID(1002, "订单状态不正确"),

    // 支付相关（2000-2999）
    PAYMENT_FAILED(2001, "支付失败"),
    PAYMENT_TIMEOUT(2002, "支付超时"),

    // Agent 相关（3000-3999）
    AGENT_TOOL_ERROR(3001, "工具调用失败"),
    AGENT_RAG_ERROR(3002, "RAG 检索异常");

    private final int code;
    private final String message;
}

// 自定义业务异常
public class BusinessException extends RuntimeException {
    private final ErrorCode errorCode;
    private final String traceId;       // 追踪 ID
    private final String module;        // 模块名
    private final Map<String, Object> context;  // 上下文参数

    public BusinessException(ErrorCode errorCode, String traceId,
            String module, Map<String, Object> context) {
        super(errorCode.getMessage());
        this.errorCode = errorCode;
        this.traceId = traceId;
        this.module = module;
        this.context = context;
    }
}

@RestControllerAdvice
public class GlobalExceptionHandler {

    private final Logger log = LoggerFactory.getLogger(this.getClass());

    @ExceptionHandler(BusinessException.class)
    public ResponseEntity<?> handleBusinessException(
            BusinessException ex, HttpServletRequest request) {

        // 记录完整异常信息
        log.error("业务异常 [TraceId:{}] [模块:{}] [错误码:{}] [用户:{}] " +
            "[路径:{}] [参数:{}] [堆栈:{}]",
            ex.getTraceId(),
            ex.getModule(),
            ex.getErrorCode().getCode(),
            getCurrentUserId(),
            request.getRequestURI(),
            ex.getContext(),
            ex.getStackTrace());

        // 返回结构化响应
        ErrorResponse response = ErrorResponse.builder()
            .code(ex.getErrorCode().getCode())
            .message(ex.getErrorCode().getMessage())
            .traceId(ex.getTraceId())
            .timestamp(LocalDateTime.now())
            .build();

        return ResponseEntity.status(HttpStatus.BAD_REQUEST).body(response);
    }

    @ExceptionHandler(Exception.class)
    public ResponseEntity<?> handleUnknownException(
            Exception ex, HttpServletRequest request) {

        String traceId = request.getHeader("X-Trace-Id");

        log.error("未知异常 [TraceId:{}] [路径:{}]", traceId,
            request.getRequestURI(), ex);

        ErrorResponse response = ErrorResponse.builder()
            .code(500)
            .message("系统内部错误")
            .traceId(traceId)
            .timestamp(LocalDateTime.now())
            .build();

        return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
            .body(response);
    }
}

// Trace ID 过滤器
@Component
public class TraceIdFilter implements Filter {

    private static final String TRACE_ID_HEADER = "X-Trace-Id";
    private static final ThreadLocal<String> TRACE_ID_HOLDER = new ThreadLocal<>();

    @Override
    public void doFilter(ServletRequest request, ServletResponse response,
            FilterChain chain) throws IOException, ServletException {
        HttpServletRequest httpRequest = (HttpServletRequest) request;

        // 生成或获取 Trace ID
        String traceId = httpRequest.getHeader(TRACE_ID_HEADER);
        if (traceId == null || traceId.isEmpty()) {
            traceId = UUID.randomUUID().toString().replace("-", "");
        }

        TRACE_ID_HOLDER.set(traceId);
        response.setHeader(TRACE_ID_HEADER, traceId);

        try {
            chain.doFilter(request, response);
        } finally {
            TRACE_ID_HOLDER.remove();
        }
    }

    public static String getCurrentTraceId() {
        return TRACE_ID_HOLDER.get();
    }
}

开发者提交代码
    ↓
┌─────────────────────────────┐
│ GitLab CI Pipeline           │
│ 1. 代码检查（SonarQube）     │
│ 2. 单元测试（JUnit）         │
│ 3. 构建打包（Maven）         │
│ 4. 构建 Docker 镜像          │
│ 5. 推送到 Harbor 仓库        │
└─────────────┬───────────────┘
              ↓
┌─────────────────────────────┐
│ Kubernetes 部署              │
│ 1. Dev 环境自动部署          │
│ 2. Test 环境手动触发         │
│ 3. Staging 环境审批后部署    │
│ 4. Production 灰度发布       │
└─────────────────────────────┘

# Deployment 配置
apiVersion: apps/v1
kind: Deployment
metadata:
  name: agent-service
  namespace: production
spec:
  replicas: 5
  selector:
    matchLabels:
      app: agent-service
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 25%
      maxUnavailable: 0  # 零宕机
  template:
    spec:
      containers:
      - name: agent-service
        image: harbor.example.com/agent-service:latest
        ports:
        - containerPort: 8080
        env:
        - name: SPRING_PROFILES_ACTIVE
          value: "prod"
        envFrom:
        - configMapRef:
            name: agent-config
        - secretRef:
            name: agent-secrets
        resources:
          requests:
            cpu: "500m"
            memory: "1Gi"
          limits:
            cpu: "2000m"
            memory: "4Gi"
        livenessProbe:
          httpGet:
            path: /actuator/health
            port: 8080
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /actuator/ready
            port: 8080
          initialDelaySeconds: 10
          periodSeconds: 5
        lifecycle:
          preStop:
            exec:
              command: ["sh", "-c", "sleep 30"]  # 优雅关闭
---
# HPA 自动扩缩容
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: agent-service-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: agent-service
  minReplicas: 3
  maxReplicas: 20
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70

# Ingress 灰度配置
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: agent-ingress
  annotations:
    nginx.ingress.kubernetes.io/canary-by-header: "canary"
    nginx.ingress.kubernetes.io/canary-weight: "10"  # 10% 流量
spec:
  rules:
  - host: agent.example.com
    http:
      paths:
      - path: /
        backend:
          service:
            name: agent-service-canary
            port:
              number: 8080
---
apiVersion: v1
kind: Service
metadata:
  name: agent-service-canary
spec:
  selector:
    app: agent-service
    version: v2  # 新版本
  ports:
  - port: 8080

┌──────────┐    Binlog     ┌──────────┐    MQ     ┌──────────┐
│  MySQL   │──────────────→│  Canal   │────────→│ RocketMQ │
│  (主库)  │               │ (监听)   │         │ (消息)   │
└──────────┘               └──────────┘         └────┬─────┘
                                                     ↓
┌──────────┐                                    ┌──────────┐
│  业务服务 │                                    │  ES Consumer
│  (写DB)  │                                    │ (消费MQ) │
└────┬─────┘                                    └────┬─────┘
     │                                               ↓
     │ 定期全量扫描                              ┌──────────┐
     └──────────────────────────────────────────→│  ElasticSearch
                                                 │ (索引)   │
                                                 └──────────┘
                                                      ↑
                                                      │ 每小时校验
                                                 ┌──────────┐
                                                 │ 数据校验  │
                                                 │ 定时任务  │
                                                 └──────────┘

# canal 配置实例.properties
canal.instance.master.address=mysql-master:3306
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal_pwd
canal.instance.filter.regex=order_db\\..*
canal.mq.exchange=canal.exchange
canal.mq.routingKey=canal.routing.key

@RocketMQMessageListener(
    topic = "canal_order_topic",
    consumerGroup = "es_sync_group"
)
public class ESSyncConsumer implements MessageListener {

    private final RestHighLevelClient esClient;

    @Override
    public void onMessage(Message msg) {
        CanalMessage canalMsg = JSON.parseObject(
            new String(msg.getBody()), CanalMessage.class
        );

        String tableName = canalMsg.getTable();
        String eventType = canalMsg.getType();  // INSERT/UPDATE/DELETE

        switch (eventType) {
            case "INSERT":
            case "UPDATE":
                // 同步到 ES（upsert 语义）
                esClient.index(IndexRequest.of(
                    i -> i.index(tableName)
                        .id(canalMsg.getId())
                        .document(canalMsg.getData())
                ));
                break;
            case "DELETE":
                // 删除 ES 文档
                esClient.delete(DeleteRequest.of(
                    tableName, canalMsg.getId()
                ));
                break;
        }
    }
}

// 解决并发更新冲突
@Entity
public class Order {
    @Version  // JPA 乐观锁
    private Long version;

    private Long id;
    private String status;
    private LocalDateTime updatedAt;
}

// ES 索引中存储版本号
public void syncToES(Order order) {
    esClient.index(IndexRequest.of(i -> {
        try {
            return i.index("orders")
                .id(String.valueOf(order.getId()))
                .document(Map.of(
                    "order_id", order.getId(),
                    "status", order.getStatus(),
                    "version", order.getVersion(),  // 版本号
                    "updated_at", order.getUpdatedAt()
                ))
                .ifSeqNoIsPresent()  // 仅当版本号更大时才更新
                .set("seq_no_primary_term",
                    getCurrentESVersion(order.getId()));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }));
}

@Scheduled(cron = "0 0 2 * * ?")  // 每天凌晨 2 点
public void fullSync() {
    List<Order> allOrders = orderRepository.findAll();

    // 分批处理，每批 500 条
    Lists.partition(allOrders, 500).forEach(batch -> {
        esClient.bulk(BulkRequest.of(b -> {
            for (Order order : batch) {
                b.index(IndexRequest.of(i ->
                    i.index("orders")
                        .id(String.valueOf(order.getId()))
                        .document(convertToDocument(order))
                ));
            }
            return b;
        }));
    });

    log.info("全量同步完成，共 {} 条", allOrders.size());
}

用户登录
    ↓
┌─────────────────────────────┐
│ 生成 Access Token (30min)   │
│ 生成 Refresh Token (7day)   │
│ 存入 Redis                   │
│ 返回两个 Token 给前端         │
└─────────────┬───────────────┘
              ↓
┌─────────────────────────────┐
│ 正常使用阶段                 │
│ 前端带 Access Token 请求     │
│ 后端校验 → 返回数据           │
│ Access Token 即将过期时      │
│ 前端自动用 Refresh Token 刷新│
└─────────────┬───────────────┘
              ↓
┌─────────────────────────────┐
│ 刷新阶段                     │
│ 校验 Refresh Token           │
│ 生成新的 Access + Refresh    │
│ 旧 Refresh Token 立即失效    │
│ 返回新的两个 Token            │
└─────────────┬───────────────┘
              ↓
┌─────────────────────────────┐
│ 安全机制                     │
│ 设备指纹绑定校验             │
│ 异常登录告警                 │
│ 管理员可强制下线              │
│ Refresh Token 过期 → 需重新登录│
└─────────────────────────────┘

// Token 服务
@Service
public class TokenService {

    private static final long ACCESS_TOKEN_EXPIRE = 30 * 60;     // 30 分钟
    private static final long REFRESH_TOKEN_EXPIRE = 7 * 24 * 3600; // 7 天

    private final JwtUtil jwtUtil;
    private final RedisTemplate redisTemplate;

    // 登录时生成双 Token
    public TokenPair generateTokenPair(Long userId, String deviceFingerprint) {
        String accessToken = jwtUtil.generate(userId, ACCESS_TOKEN_EXPIRE);
        String refreshToken = jwtUtil.generate(
            userId + ":refresh", REFRESH_TOKEN_EXPIRE
        );

        // 存入 Redis
        redisTemplate.opsForValue().set(
            "token:" + userId,
            accessToken,
            ACCESS_TOKEN_EXPIRE, TimeUnit.SECONDS
        );
        redisTemplate.opsForValue().set(
            "refresh:" + userId,
            refreshToken + ":" + deviceFingerprint,
            REFRESH_TOKEN_EXPIRE, TimeUnit.SECONDS
        );

        return new TokenPair(accessToken, refreshToken);
    }

    // 刷新 Token（Token 旋转）
    public TokenPair refresh(String refreshToken, String deviceFingerprint) {
        // 1. 校验 Refresh Token
        Long userId = jwtUtil.parseUserId(refreshToken);
        String stored = redisTemplate.opsForValue()
            .get("refresh:" + userId);

        if (stored == null) {
            throw new AuthException("Refresh Token 已失效");
        }

        // 2. 设备指纹校验
        String storedFingerprint = stored.split(":")[1];
        if (!storedFingerprint.equals(deviceFingerprint)) {
            // 设备不一致，告警
            alertService.alertAbnormalLogin(userId);
            throw new AuthException("设备不一致");
        }

        // 3. 旧 Refresh Token 立即失效（防止重放）
        redisTemplate.delete("refresh:" + userId);

        // 4. 生成新 Token 对
        return generateTokenPair(userId, deviceFingerprint);
    }

    // 强制下线
    public void forceLogout(Long userId) {
        redisTemplate.delete("token:" + userId);
        redisTemplate.delete("refresh:" + userId);
    }
}

JWT Token 结构：
┌─────────────────────────────────────────────────┐
│ Header.Payload.Signature                        │
├─────────────────────────────────────────────────┤
│                                                  │
│ Header（Header）:                                │
│ {                                                │
│   "alg": "HS256",    // 签名算法                │
│   "typ": "JWT"       // Token 类型              │
│ }                                                │
│                                                  │
│ Payload（载荷）:                                 │
│ {                                                │
│   "sub": "1234567890",  // 主题（用户ID）        │
│   "name": "John Doe",   // 用户名               │
│   "iat": 1516239022,    // 签发时间             │
│   "exp": 1516240000     // 过期时间             │
│ }                                                │
│                                                  │
│ Signature（签名）:                              │
│ HMACSHA256(                                     │
│   base64(Header) + "." + base64(Payload),       │
│   secretKey                                     │
│ )                                                │
│                                                  │
└─────────────────────────────────────────────────┘

客户端                          服务端
  │                              │
  │ 1. POST /api/auth/login      │
  │    {username, password}      │
  │─────────────────────────────→│
  │                              │
  │                              │ 2. 验证用户名密码
  │                              │    - 查询用户
  │                              │    - BCrypt 校验密码
  │                              │    - 检查账号状态
  │                              │
  │                              │ 3. 生成 JWT
  │                              │    - Header: {"alg":"HS256","typ":"JWT"}
  │                              │    - Payload: {sub, userId, role, exp}
  │                              │    - Signature: HMACSHA256(header+payload)
  │                              │
  │ 4. 返回 Token                │
  │    {accessToken, refreshToken}│
  │←─────────────────────────────│
  │                              │
  │ 5. GET /api/orders           │
  │    Authorization: Bearer <JWT>│
  │─────────────────────────────→│
  │                              │
  │                              │ 6. 校验 JWT
  │                              │    - 解析 Header.Payload.Signature
  │                              │    - 验证签名（确保未篡改）
  │                              │    - 检查 exp 是否过期
  │                              │    - 检查 Redis 黑名单
  │                              │    - 提取 userId、role
  │                              │
  │                              │ 7. 权限校验
  │                              │    - 检查角色是否有权访问
  │                              │    - 数据权限过滤
  │                              │
  │ 8. 返回数据                  │
  │←─────────────────────────────│

// JWT 工具类
@Component
public class JwtUtil {

    private static final String SECRET_KEY = "your-256-bit-secret-key-here";
    private static final long ACCESS_TOKEN_EXPIRE = 30 * 60;     // 30 分钟
    private static final long REFRESH_TOKEN_EXPIRE = 7 * 24 * 3600; // 7 天

    // 生成 Token
    public String generateAccessToken(Long userId, String role) {
        return Jwts.builder()
            .setHeaderParam("alg", "HS256")
            .setHeaderParam("typ", "JWT")
            .setSubject(String.valueOf(userId))
            .claim("role", role)
            .claim("type", "access")
            .setIssuedAt(new Date())
            .setExpiration(new Date(System.currentTimeMillis()
                + ACCESS_TOKEN_EXPIRE * 1000))
            .signWith(Keys.hmacShaKeyFor(SECRET_KEY.getBytes()),
                SignatureAlgorithm.HS256)
            .compact();
    }

    // 解析和验证 Token
    public Claims parseToken(String token) {
        return Jwts.parserBuilder()
            .setSigningKey(Keys.hmacShaKeyFor(SECRET_KEY.getBytes()))
            .build()
            .parseClaimsJws(token)
            .getBody();
    }

    // 验证 Token 并提取用户信息
    public UserInfo validateAndGetUser(String token) {
        Claims claims = parseToken(token);

        // 检查过期
        if (claims.getExpiration().before(new Date())) {
            throw new JwtException("Token 已过期");
        }

        return new UserInfo(
            Long.parseLong(claims.getSubject()),
            claims.get("role", String.class),
            claims.get("type", String.class)
        );
    }
}

@Component
public class JwtAuthenticationFilter extends OncePerRequestFilter {

    private final JwtUtil jwtUtil;
    private final UserDetailsService userDetailsService;

    @Override
    protected void doFilterInternal(HttpServletRequest request,
            HttpServletResponse response, FilterChain filterChain)
            throws ServletException, IOException {

        String authHeader = request.getHeader("Authorization");

        if (authHeader == null || !authHeader.startsWith("Bearer ")) {
            filterChain.doFilter(request, response);
            return;
        }

        String token = authHeader.substring(7);

        try {
            UserInfo userInfo = jwtUtil.validateAndGetUser(token);

            UsernamePasswordAuthenticationToken auth =
                new UsernamePasswordAuthenticationToken(
                    userInfo, null,
                    getAuthorities(userInfo.getRole())
                );

            SecurityContextHolder.getContext().setAuthentication(auth);

        } catch (JwtException e) {
            // Token 无效，返回 401
            response.setStatus(HttpStatus.UNAUTHORIZED.value());
            response.setContentType("application/json");
            response.getWriter().write(
                "{\"error\":\"Invalid or expired token\"}"
            );
            return;
        }

        filterChain.doFilter(request, response);
    }
}

// JWT 黑名单 + Token 版本号
@Service
public class JwtSecurityService {

    private final RedisTemplate redisTemplate;

    // 强制下线：将 Token 加入黑名单
    public void blacklistToken(String token, long remainingTime) {
        redisTemplate.opsForValue().set(
            "jwt:blacklist:" + token,
            "1",
            remainingTime, TimeUnit.SECONDS
        );
    }

    // 检查 Token 是否在黑名单
    public boolean isBlacklisted(String token) {
        return Boolean.TRUE.equals(
            redisTemplate.hasKey("jwt:blacklist:" + token)
        );
    }

    // Token 版本号：支持修改密码后所有 Token 失效
    public void incrementTokenVersion(Long userId) {
        String key = "jwt:version:" + userId;
        redisTemplate.opsForValue().increment(key);
    }

    // 验证 Token 版本号
    public boolean verifyTokenVersion(Long userId, long tokenVersion) {
        String key = "jwt:version:" + userId;
        String currentVersion = redisTemplate.opsForValue().get(key);
        if (currentVersion == null) return true; // 无版本号，允许访问
        return Long.parseLong(currentVersion) == tokenVersion;
    }
}