基于Spring Cloud Alibaba与Kubernetes的Java微服务架构实践：从网关到链路追踪全栈解析

@Configuration
public class GrayRouteConfig {
    @Bean
    public RouteLocator grayRouteLocator(RouteLocatorBuilder builder) {
        return builder.routes()
            .route("gray_service", r -> r
                .header("version", "v2")   // 携带v2头走灰度
                .and()
                .weight("service-a", 10)   // 10%流量随机走灰度
                .uri("lb://service-a-gray"))
            .route("stable_service", r -> r
                .alwaysTrue()
                .uri("lb://service-a"))
            .build();
    }
}

@Component
public class TraceIdFilter implements GlobalFilter, Ordered {
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        String traceId = exchange.getRequest().getHeaders().getFirst("X-Trace-Id");
        if (StringUtils.isEmpty(traceId)) {
            traceId = UUID.randomUUID().toString().replace("-", "");
        }
        // 写入MDC，同时向下游传递
        MDC.put("traceId", traceId);
        ServerHttpRequest mutated = exchange.getRequest().mutate()
                .header("X-Trace-Id", traceId)
                .build();
        return chain.filter(exchange.mutate().request(mutated).build())
                .doFinally(signal -> MDC.clear());
    }
    @Override
    public int getOrder() { return -1000; }
}

@Component
public class NacosConfigListener implements ApplicationListener<EnvironmentChangeEvent> {
    private static final Map<String, Object> CONFIG_SNAPSHOT = new ConcurrentHashMap<>();
    
    @Override
    public void onApplicationEvent(EnvironmentChangeEvent event) {
        Set<String> keys = event.getKeys();
        for (String key : keys) {
            // 变更前备份
            CONFIG_SNAPSHOT.put(key, environment.getProperty(key));
            // 触发业务回调，如线程池参数调整
            if (key.startsWith("thread.pool.")) {
                ThreadPoolManager.refresh(key, environment.getProperty(key, Integer.class));
            }
        }
    }
    
    // 回滚API
    public void rollback(String key) {
        if (CONFIG_SNAPSHOT.containsKey(key)) {
            // 通过Nacos OpenAPI 推送旧值
            nacosClient.publishConfig(key, DEFAULT_GROUP, CONFIG_SNAPSHOT.get(key).toString());
        }
    }
}

@ConfigurationProperties(prefix = "thread.pool")
@Data
public class ThreadPoolProperties {
    private int coreSize = 10;
    private int maxSize = 50;
    private int queueCapacity = 200;
}

@Bean
@RefreshScope
public ThreadPoolExecutor bizExecutor(ThreadPoolProperties props) {
    return new ThreadPoolExecutor(
        props.getCoreSize(),
        props.getMaxSize(),
        60L, TimeUnit.SECONDS,
        new LinkedBlockingQueue<>(props.getQueueCapacity()),
        new NamedThreadFactory("biz-"),
        new CallerRunsPolicy()  // 背压保护
    );
}

@LocalTCC
public interface StockTccAction {
    @TwoPhaseBusinessAction(name = "deductStock", commitMethod = "commit", rollbackMethod = "rollback")
    boolean tryDeduct(@BusinessActionContextParameter(paramName = "skuId") Long skuId,
                      @BusinessActionContextParameter(paramName = "count") Integer count,
                      BusinessActionContext context);
    
    boolean commit(BusinessActionContext context);
    boolean rollback(BusinessActionContext context);
}

@Component
public class StockTccActionImpl implements StockTccAction {
    @Autowired
    private TccTransactionLogMapper logMapper;
    
    @Override
    public boolean tryDeduct(Long skuId, Integer count, BusinessActionContext context) {
        String xid = context.getXid();
        // 先查日志，若已回滚则直接返回false防止悬挂
        TccLog log = logMapper.selectByXid(xid);
        if (log != null && log.getStatus() == TccStatus.ROLLBACKED) {
            return false; // 空回滚
        }
        // 扣减库存（Redis预扣 + DB异步落盘）
        int affected = stockService.tryFreeze(skuId, count);
        if (affected > 0) {
            logMapper.insert(new TccLog(xid, "TRY", TccStatus.TRYING));
            return true;
        }
        return false;
    }
    
    @Override
    public boolean commit(BusinessActionContext context) {
        TccLog log = logMapper.selectByXid(context.getXid());
        if (log == null || log.getStatus() == TccStatus.COMMITTED) {
            return true; // 幂等
        }
        // 真正扣减DB
        stockService.confirmDeduct(context.getActionContext("skuId"));
        log.setStatus(TccStatus.COMMITTED);
        logMapper.updateById(log);
        return true;
    }
}

<Appenders>
    <Console name="Console" target="SYSTEM_OUT">
        <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss.SSS} [%traceId] [%tid] %-5level %logger{36} - %msg%n"/>
    </Console>
    <Elasticsearch name="esAppender" 
                   indexName="app-logs-${date:yyyy-MM}"
                   servers="es-cluster:9200">
        <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss.SSS} %traceId %tid %level %logger %msg"/>
    </Elasticsearch>
</Appenders>

@Aspect
@Component
public class BizSpanAspect {
    @Around("@annotation(io.swagger.annotations.ApiOperation)")
    public Object around(ProceedingJoinPoint pjp) throws Throwable {
        Tracer tracer = GlobalTracer.get();
        Span span = tracer.activeSpan();
        if (span != null) {
            span.setTag("biz.method", pjp.getSignature().getName());
            // 提取订单号等
            Object[] args = pjp.getArgs();
            for (Object arg : args) {
                if (arg instanceof OrderRequest) {
                    span.setTag("order.id", ((OrderRequest) arg).getOrderId());
                }
            }
        }
        return pjp.proceed();
    }
}

lifecycle:
  preStop:
    exec:
      command: ["/bin/sh", "-c", "sleep 15 && /app/shutdown.sh"]
readinessProbe:
  httpGet:
    path: /actuator/health/readiness
    port: 8080
  initialDelaySeconds: 30
  periodSeconds: 5
  failureThreshold: 3

apiVersion: v1
kind: Service
metadata:
  name: biz-service
spec:
  externalTrafficPolicy: Local
  publishNotReadyAddresses: false  # 关键：就绪前不注册Endpoint
  ports:
  - port: 8080
  selector:
    app: biz-service

@Component
public class GracefulShutdown implements SmartLifecycle {
    private volatile boolean running = true;
    private final ThreadPoolExecutor executor;
    
    @PreDestroy
    public void shutdown() {
        running = false;
        executor.shutdown();
        try {
            if (!executor.awaitTermination(20, TimeUnit.SECONDS)) {
                executor.shutdownNow();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        // 关闭数据库连接池
        HikariDataSource ds = (HikariDataSource) dataSource;
        ds.close();
    }
}

-Xms4g -Xmx4g -XX:+UseG1GC -XX:MaxGCPauseMillis=200 
-XX:G1HeapRegionSize=16m -XX:InitiatingHeapOccupancyPercent=45 
-XX:+PrintGCDetails -Xloggc:/logs/gc.log 
-XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/dump

public class AdaptiveRejectPolicy implements RejectedExecutionHandler {
    private volatile RejectedExecutionHandler current = new CallerRunsPolicy();
    
    @Scheduled(fixedDelay = 5000)
    public void adjust() {
        double load = ManagementFactory.getOperatingSystemMXBean().getSystemLoadAverage();
        if (load > 0.7 * Runtime.getRuntime().availableProcessors()) {
            current = new AbortPolicy();
            alarmService.send("线程池切换为拒绝抛出，当前负载:" + load);
        } else {
            current = new CallerRunsPolicy();
        }
    }
}