Spring Cloud微服务¶

@SpringBootApplication
@EnableDiscoveryClient  // Spring Cloud标准注解，兼容多种注册中心
public class UserServiceApplication {
    public static void main(String[] args) {
        SpringApplication.run(UserServiceApplication.class, args);
    }
}

// ============ 声明式Feign客户端 ============
@FeignClient(name = "product-service", fallbackFactory = ProductClientFallback.class)
public interface ProductClient {

    @GetMapping("/api/products/{id}")
    ProductDTO getProduct(@PathVariable("id") Long id);

    @GetMapping("/api/products/batch")
    List<ProductDTO> batchGetProducts(@RequestParam("ids") List<Long> ids);
}

// ============ 降级工厂 ============
@Component
public class ProductClientFallback implements FallbackFactory<ProductClient> {
    @Override
    public ProductClient create(Throwable cause) {
        return new ProductClient() {
            @Override
            public ProductDTO getProduct(Long id) {
                // 降级逻辑：返回兜底数据
                return ProductDTO.builder()
                    .id(id)
                    .name("降级商品")
                    .price(BigDecimal.ZERO)
                    .build();
            }

            @Override
            public List<ProductDTO> batchGetProducts(List<Long> ids) {
                return Collections.emptyList();
            }
        };
    }
}

// 临时实例（AP模式，默认）— 适合微服务
// 心跳上报，Nacos不主动检查，注册快速
spring.cloud.nacos.discovery.ephemeral=true

// 持久实例（CP模式）— 适合基础设施服务（如数据库代理）
// Nacos主动健康检查，适合不能主动上报心跳的服务
spring.cloud.nacos.discovery.ephemeral=false

/**
 * JWT鉴权全局过滤器
 */
@Component
@Order(-1)  // 优先级最高
public class AuthGlobalFilter implements GlobalFilter {

    private static final List<String> WHITE_LIST = List.of(
        "/api/auth/login",
        "/api/auth/register",
        "/api/products/public"
    );

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        ServerHttpRequest request = exchange.getRequest();
        String path = request.getPath().value();

        // 白名单放行
        if (WHITE_LIST.stream().anyMatch(path::startsWith)) {
            return chain.filter(exchange);
        }

        // 提取Token
        String token = request.getHeaders().getFirst("Authorization");
        if (token == null || !token.startsWith("Bearer ")) {
            return unauthorized(exchange, "Missing or invalid token");
        }

        // 验证Token
        try {
            String jwt = token.substring(7);
            Claims claims = JwtUtil.parseToken(jwt);
            String userId = claims.getSubject();

            // 将用户信息传递给下游服务
            ServerHttpRequest mutatedRequest = request.mutate()
                .header("X-User-Id", userId)
                .header("X-User-Role", claims.get("role", String.class))
                .build();

            return chain.filter(exchange.mutate().request(mutatedRequest).build());
        } catch (ExpiredJwtException e) {
            return unauthorized(exchange, "Token expired");
        } catch (Exception e) {
            return unauthorized(exchange, "Invalid token");
        }
    }

    private Mono<Void> unauthorized(ServerWebExchange exchange, String message) {
        ServerHttpResponse response = exchange.getResponse();
        response.setStatusCode(HttpStatus.UNAUTHORIZED);
        response.getHeaders().setContentType(MediaType.APPLICATION_JSON);
        String body = """
            {"code": 401, "message": "%s"}
            """.formatted(message);
        DataBuffer buffer = response.bufferFactory().wrap(body.getBytes(StandardCharsets.UTF_8));
        return response.writeWith(Mono.just(buffer));
    }
}

/**
 * 基于Redis的请求速率限流
 */
@Configuration
public class RateLimitConfig {

    /**
     * 按IP限流的KeyResolver
     */
    @Bean
    public KeyResolver ipKeyResolver() {
        return exchange -> Mono.just(
            Objects.requireNonNull(exchange.getRequest().getRemoteAddress())
                .getAddress().getHostAddress()
        );
    }
}

/**
 * 自定义负载均衡配置
 * 注意：不要添加 @Configuration，避免被全局扫描
 */
public class CustomLoadBalancerConfig {

    /**
     * 轮询策略（默认）
     */
    @Bean
    public ReactorLoadBalancer<ServiceInstance> roundRobinLoadBalancer(
            Environment environment,
            LoadBalancerClientFactory loadBalancerClientFactory) {
        String name = environment.getProperty(LoadBalancerClientFactory.PROPERTY_NAME);
        return new RoundRobinLoadBalancer(
            loadBalancerClientFactory.getLazyProvider(name, ServiceInstanceListSupplier.class),
            name
        );
    }

    /**
     * 随机策略
     */
    @Bean
    public ReactorLoadBalancer<ServiceInstance> randomLoadBalancer(
            Environment environment,
            LoadBalancerClientFactory loadBalancerClientFactory) {
        String name = environment.getProperty(LoadBalancerClientFactory.PROPERTY_NAME);
        return new RandomLoadBalancer(
            loadBalancerClientFactory.getLazyProvider(name, ServiceInstanceListSupplier.class),
            name
        );
    }
}

// ============ 指定服务使用特定负载均衡策略 ============
@Configuration
@LoadBalancerClients({
    @LoadBalancerClient(name = "user-service", configuration = CustomLoadBalancerConfig.class),
    @LoadBalancerClient(name = "order-service", configuration = CustomLoadBalancerConfig.class)
})
public class LoadBalancerAutoConfig {
}

/**
 * Nacos权重负载均衡器 — 结合Nacos控制台动态调权
 */
public class NacosWeightLoadBalancer implements ReactorServiceInstanceLoadBalancer {

    private final String serviceId;
    private final ObjectProvider<ServiceInstanceListSupplier> supplierProvider;

    public NacosWeightLoadBalancer(
            ObjectProvider<ServiceInstanceListSupplier> supplierProvider,
            String serviceId) {
        this.supplierProvider = supplierProvider;
        this.serviceId = serviceId;
    }

    @Override
    public Mono<Response<ServiceInstance>> choose(Request request) {
        return supplierProvider.getIfAvailable()
            .get()
            .next()
            .map(this::selectByWeight);
    }

    private Response<ServiceInstance> selectByWeight(List<ServiceInstance> instances) {
        if (instances.isEmpty()) {
            return new EmptyResponse();
        }

        // 使用Nacos权重进行随机选择
        List<Pair<ServiceInstance, Double>> weightedInstances = instances.stream()
            .map(inst -> {
                double weight = Double.parseDouble(
                    inst.getMetadata().getOrDefault("nacos.weight", "1.0"));
                return Pair.of(inst, weight);
            })
            .toList();

        double totalWeight = weightedInstances.stream()
            .mapToDouble(Pair::getRight).sum();
        double random = ThreadLocalRandom.current().nextDouble(totalWeight);

        double cumulative = 0;
        for (Pair<ServiceInstance, Double> pair : weightedInstances) {
            cumulative += pair.getRight();
            if (random < cumulative) {
                return new DefaultResponse(pair.getLeft());
            }
        }
        return new DefaultResponse(instances.get(0));
    }
}

/**
 * 动态配置示例：降级开关、限流阈值等运行时可调整参数
 */
@Component
@RefreshScope  // Spring Cloud原生注解，配置变更时重新创建Bean
public class DynamicConfigProperties {

    @Value("${order.max-amount:99999}")
    private BigDecimal maxOrderAmount;

    @Value("${feature.toggle.new-checkout:false}")
    private boolean newCheckoutEnabled;

    @Value("${rate-limit.qps:100}")
    private int rateLimitQps;

    // Getters ...
    public BigDecimal getMaxOrderAmount() { return maxOrderAmount; }
    public boolean isNewCheckoutEnabled() { return newCheckoutEnabled; }
    public int getRateLimitQps() { return rateLimitQps; }
}

/**
 * 也可以使用 @ConfigurationProperties 方式
 */
@Component
@ConfigurationProperties(prefix = "order")
@RefreshScope
public class OrderConfig {
    private BigDecimal maxAmount;
    private int timeout;
    private RetryConfig retry;

    @Data
    public static class RetryConfig {
        private int maxRetries;
        private long delay;
    }

    // Getters & Setters ...
}

/**
 * 敏感配置加密 — 结合Jasypt
 */
// 1. 引入依赖
// com.github.ulisesbocchio:jasypt-spring-boot-starter:3.0.5

// 2. 配置
// jasypt.encryptor.password=your-secret-key

// 3. 加密值使用 ENC() 包裹
// spring.datasource.password=ENC(加密后的密文)

@Configuration
public class JasyptConfig {
    @Bean("jasyptStringEncryptor")
    public StringEncryptor stringEncryptor() {
        PooledPBEStringEncryptor encryptor = new PooledPBEStringEncryptor();
        SimpleStringPBEConfig config = new SimpleStringPBEConfig();
        config.setPassword(System.getenv("JASYPT_PASSWORD")); // 从环境变量获取密钥
        config.setAlgorithm("PBEWITHHMACSHA512ANDAES_256");
        config.setKeyObtentionIterations("1000");
        config.setPoolSize("1");
        config.setSaltGeneratorClassName("org.jasypt.salt.RandomSaltGenerator");
        encryptor.setConfig(config);
        return encryptor;
    }
}

/**
 * 注解方式定义Sentinel资源 + 降级
 */
@Service
public class OrderService {

    /**
     * @SentinelResource 定义资源，配置降级和异常处理
     *   - value: 资源名称
     *   - blockHandler: 被Sentinel限流/降级时的处理方法
     *   - fallback: 业务异常时的降级方法
     */
    @SentinelResource(
        value = "createOrder",
        blockHandler = "createOrderBlockHandler",
        fallback = "createOrderFallback"
    )
    public OrderDTO createOrder(CreateOrderRequest request) {
        // 正常业务逻辑
        ProductDTO product = productClient.getProduct(request.getProductId());
        if (product.getStock() < request.getQuantity()) {
            throw new BusinessException("库存不足");
        }
        // ... 创建订单
        return orderDTO;
    }

    /**
     * 被Sentinel限流/熔断时调用
     * 参数列表必须与原方法一致 + 多一个BlockException
     */
    public OrderDTO createOrderBlockHandler(CreateOrderRequest request, BlockException ex) {
        if (ex instanceof FlowException) {
            throw new ApiException(429, "下单太快了，请稍后重试");
        }
        if (ex instanceof DegradeException) {
            throw new ApiException(503, "订单服务暂时不可用，请稍后重试");
        }
        throw new ApiException(429, "请求被限流: " + ex.getClass().getSimpleName());
    }

    /**
     * 业务异常降级
     */
    public OrderDTO createOrderFallback(CreateOrderRequest request, Throwable ex) {
        log.error("创建订单降级处理, request={}", request, ex);
        // 写入消息队列延迟处理
        mqTemplate.asyncSend("ORDER_FALLBACK", request);
        return OrderDTO.builder()
            .status("PENDING")
            .message("订单已提交，稍后处理")
            .build();
    }
}

/**
 * 热点参数限流 — 针对频繁访问的特定参数值限流
 * 场景：秒杀活动中，对热门商品ID限流
 */
@SentinelResource(
    value = "getProduct",
    blockHandler = "getProductBlockHandler"
)
public ProductDTO getProduct(Long productId) {
    return productRepository.findById(productId)
        .map(this::convertToDTO)
        .orElseThrow(() -> new NotFoundException("商品不存在: " + productId));
}

/**
 * 编程方式配置热点参数限流规则
 */
@PostConstruct
public void initHotParamRules() {
    ParamFlowRule rule = new ParamFlowRule();
    rule.setResource("getProduct");
    rule.setParamIdx(0);                  // 第0个参数（productId）
    rule.setGrade(RuleConstant.FLOW_GRADE_QPS);
    rule.setCount(100);                   // 默认每个参数值QPS上限100

    // 特殊值例外项：热门商品ID限流更严格
    ParamFlowItem item1 = new ParamFlowItem()
        .setObject("1001")                // 热门商品ID
        .setClassType(long.class.getName())
        .setCount(20);                    // 该商品QPS上限20
    ParamFlowItem item2 = new ParamFlowItem()
        .setObject("1002")
        .setClassType(long.class.getName())
        .setCount(20);

    rule.setParamFlowItemList(List.of(item1, item2));
    ParamFlowRuleManager.loadRules(List.of(rule));
}

/**
 * 编程方式配置熔断降级规则
 */
@PostConstruct
public void initDegradeRules() {
    List<DegradeRule> rules = new ArrayList<>();

    // 策略1：慢调用比例 — RT超过200ms的请求比例超过50%时熔断
    DegradeRule slowCallRule = new DegradeRule("createOrder")
        .setGrade(CircuitBreakerStrategy.SLOW_REQUEST_RATIO.getType())
        .setCount(0.5)           // 慢调用比例阈值50%
        .setSlowRatioThreshold(200)  // 慢调用RT阈值200ms
        .setTimeWindow(30)       // 熔断持续30秒
        .setMinRequestAmount(10) // 最小统计请求数
        .setStatIntervalMs(10000); // 统计时长10秒

    // 策略2：异常比例 — 异常比例超过30%时熔断
    DegradeRule errorRatioRule = new DegradeRule("createOrder")
        .setGrade(CircuitBreakerStrategy.ERROR_RATIO.getType())
        .setCount(0.3)           // 异常比例30%
        .setTimeWindow(60)       // 熔断持续60秒
        .setMinRequestAmount(10)
        .setStatIntervalMs(10000);

    // 策略3：异常数 — 异常数超过5时熔断
    DegradeRule errorCountRule = new DegradeRule("queryOrder")
        .setGrade(CircuitBreakerStrategy.ERROR_COUNT.getType())
        .setCount(5)             // 异常数5
        .setTimeWindow(60)
        .setMinRequestAmount(5)
        .setStatIntervalMs(60000);

    rules.addAll(List.of(slowCallRule, errorRatioRule, errorCountRule));
    DegradeRuleManager.loadRules(rules);
}

/**
 * AT模式 — 下单扣库存扣余额（跨服务事务）
 * 只需在发起方加 @GlobalTransactional
 */
@Service
public class OrderService {

    @Autowired private OrderRepository orderRepository;
    @Autowired private ProductClient productClient;
    @Autowired private AccountClient accountClient;

    @GlobalTransactional(name = "create-order", rollbackFor = Exception.class)
    public OrderDTO createOrder(CreateOrderRequest request) {
        // Step 1: 创建订单
        Order order = Order.builder()
            .userId(request.getUserId())
            .productId(request.getProductId())
            .quantity(request.getQuantity())
            .totalAmount(request.getTotalAmount())
            .status(OrderStatus.PENDING)
            .build();
        orderRepository.save(order);

        // Step 2: 扣减库存（远程调用product-service）
        productClient.deductStock(request.getProductId(), request.getQuantity());

        // Step 3: 扣减余额（远程调用account-service）
        accountClient.deductBalance(request.getUserId(), request.getTotalAmount());

        // Step 4: 更新订单状态
        order.setStatus(OrderStatus.CONFIRMED);
        orderRepository.save(order);

        // 任何一步失败，Seata自动回滚所有分支事务
        return convertToDTO(order);
    }
}

/**
 * TCC模式 — 库存扣减
 * 需要手动实现Try/Confirm/Cancel三个方法
 */
@LocalTCC
public interface StockTccAction {

    /**
     * Try阶段：冻结库存
     */
    @TwoPhaseBusinessAction(
        name = "deductStock",
        commitMethod = "confirm",
        rollbackMethod = "cancel"
    )
    boolean tryDeduct(
        @BusinessActionContextParameter(paramName = "productId") Long productId,
        @BusinessActionContextParameter(paramName = "quantity") int quantity
    );

    /**
     * Confirm阶段：确认扣减（释放冻结量）
     */
    boolean confirm(BusinessActionContext context);

    /**
     * Cancel阶段：取消冻结，恢复库存
     */
    boolean cancel(BusinessActionContext context);
}

@Service
public class StockTccActionImpl implements StockTccAction {

    @Autowired private StockRepository stockRepository;

    @Override
    @Transactional
    public boolean tryDeduct(Long productId, int quantity) {
        // 冻结库存：available - quantity, frozen + quantity
        int updated = stockRepository.freezeStock(productId, quantity);
        if (updated == 0) {
            throw new BusinessException("库存不足");
        }
        return true;
    }

    @Override
    @Transactional
    public boolean confirm(BusinessActionContext context) {
        Long productId = context.getActionContext("productId", Long.class);
        int quantity = context.getActionContext("quantity", Integer.class);
        // 确认扣减：frozen - quantity
        stockRepository.confirmDeduct(productId, quantity);
        return true;
    }

    @Override
    @Transactional
    public boolean cancel(BusinessActionContext context) {
        Long productId = context.getActionContext("productId", Long.class);
        int quantity = context.getActionContext("quantity", Integer.class);
        // 取消冻结：available + quantity, frozen - quantity
        stockRepository.cancelFreeze(productId, quantity);
        return true;
    }
}

/**
 * Saga模式 — 状态机编排方式
 * 通过JSON定义状态机流转
 */
// saga-statemachine.json（Seata Saga状态机定义）
/*
{
  "Name": "createOrderSaga",
  "Comment": "创建订单Saga",
  "StartState": "CreateOrder",
  "States": {
    "CreateOrder": {
      "Type": "ServiceTask",
      "ServiceName": "orderService",
      "ServiceMethod": "createOrder",
      "CompensateState": "CompensateCreateOrder",
      "Next": "DeductStock"
    },
    "DeductStock": {
      "Type": "ServiceTask",
      "ServiceName": "stockService",
      "ServiceMethod": "deductStock",
      "CompensateState": "CompensateDeductStock",
      "Next": "DeductBalance"
    },
    "DeductBalance": {
      "Type": "ServiceTask",
      "ServiceName": "accountService",
      "ServiceMethod": "deductBalance",
      "CompensateState": "CompensateDeductBalance",
      "Next": "Succeed"
    },
    "Succeed": { "Type": "Succeed" },
    "CompensateCreateOrder": { ... },
    "CompensateDeductStock": { ... },
    "CompensateDeductBalance": { ... }
  }
}
*/

// AT vs TCC vs Saga 对比
// | 模式 | 侵入性 | 性能    | 适用场景           |
// |------|--------|---------|-------------------|
// | AT   | 无侵入  | 中等    | 中低并发的CRUD      |
// | TCC  | 高侵入  | 高      | 高并发资金类操作     |
// | Saga | 中侵入  | 中等    | 长事务/多步骤流程    |

@Service
public class PaymentService {

    private final Tracer tracer;

    public PaymentService(Tracer tracer) {
        this.tracer = tracer;
    }

    public PaymentResult processPayment(PaymentRequest request) {
        // 创建自定义Span
        Span span = tracer.nextSpan().name("process-payment").start();
        try (Tracer.SpanInScope ws = tracer.withSpan(span)) {
            span.tag("payment.method", request.getMethod());
            span.tag("payment.amount", request.getAmount().toString());
            span.tag("payment.orderId", request.getOrderId());

            // 调用第三方支付
            PaymentResult result = callThirdPartyPayment(request);

            span.tag("payment.status", result.getStatus());
            span.event("payment-completed");
            return result;
        } catch (Exception e) {
            span.error(e);
            throw e;
        } finally {
            span.end();
        }
    }
}

/**
 * Spring Cloud Stream 函数式消息处理
 * Spring Cloud 2022.x+ 推荐使用函数式方式
 */
@Configuration
public class MessageFunctions {

    /**
     * 消费者：处理订单创建事件
     */
    @Bean
    public Consumer<OrderCreatedEvent> orderCreated() {
        return event -> {
            log.info("收到订单创建事件: orderId={}, userId={}",
                event.getOrderId(), event.getUserId());
            // 发送通知、积分处理等
            notificationService.sendOrderConfirmation(event);
            pointService.addPoints(event.getUserId(), event.getAmount());
        };
    }

    /**
     * 供应者：定时生成报表事件
     */
    @Bean
    public Supplier<ReportEvent> reportGenerator() {
        return () -> {
            ReportEvent event = new ReportEvent();
            event.setTimestamp(Instant.now());
            event.setType("DAILY_SUMMARY");
            return event;
        };
    }

    /**
     * 处理器：消息转换（输入 → 处理 → 输出）
     */
    @Bean
    public Function<PaymentEvent, NotificationEvent> paymentProcessor() {
        return paymentEvent -> {
            log.info("处理支付事件: {}", paymentEvent);
            NotificationEvent notification = new NotificationEvent();
            notification.setUserId(paymentEvent.getUserId());
            notification.setMessage("支付成功，金额: " + paymentEvent.getAmount());
            notification.setChannel("SMS");
            return notification;
        };
    }
}

@Service
public class OrderEventPublisher {

    @Autowired
    private StreamBridge streamBridge;

    /**
     * 手动发送消息到指定Topic
     */
    public void publishOrderCreatedEvent(Order order) {
        OrderCreatedEvent event = OrderCreatedEvent.builder()
            .orderId(order.getId())
            .userId(order.getUserId())
            .amount(order.getTotalAmount())
            .timestamp(Instant.now())
            .build();

        // 发送到指定绑定
        boolean sent = streamBridge.send("orderCreated-out-0", event);
        log.info("订单创建事件发送{}: orderId={}", sent ? "成功" : "失败", order.getId());
    }

    /**
     * 发送延迟消息（RocketMQ特性）
     */
    public void publishDelayedEvent(String orderId) {
        Message<String> message = MessageBuilder.withPayload(orderId)
            .setHeader("DELAY_LEVEL", 16)  // RocketMQ延迟等级（约2小时）
            .build();
        streamBridge.send("order-timeout-out-0", message);
    }
}