调试指南¶

@Component
public class TraceIdFilter implements WebFilter {

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
        String traceId = UUID.randomUUID().toString().substring(0, 8);

        return chain.filter(exchange)
            .contextWrite(Context.of("traceId", traceId))
            .doOnEach(ReactiveUtils.addToMDC("traceId", traceId));
    }
}

@Service
public class LoadBalancerService {
    private static final Logger log = LoggerFactory.getLogger(LoadBalancerService.class);

    public ServiceInstance selectInstance(String serviceType, String clientInfo) {
        log.debug("Selecting instance for service: {}, client: {}", serviceType, clientInfo);

        List<ServiceInstance> instances = getAvailableInstances(serviceType);
        log.debug("Available instances: {}", instances.size());

        ServiceInstance selected = loadBalancer.selectInstance(instances, clientInfo);
        log.info("Selected instance: {} for service: {}", 
                selected != null ? selected.getName() : "null", serviceType);

        return selected;
    }
}

@Component
public class ThreadMonitor {
    private static final Logger log = LoggerFactory.getLogger(ThreadMonitor.class);

    @Scheduled(fixedRate = 60000) // 每分钟检查一次
    public void monitorThreads() {
        ThreadMXBean threadBean = ManagementFactory.getThreadMXBean();

        long[] deadlockedThreads = threadBean.findDeadlockedThreads();
        if (deadlockedThreads != null) {
            log.error("检测到死锁线程: {}", Arrays.toString(deadlockedThreads));
        }

        int threadCount = threadBean.getThreadCount();
        int peakThreadCount = threadBean.getPeakThreadCount();

        log.debug("当前线程数: {}, 峰值线程数: {}", threadCount, peakThreadCount);

        if (threadCount > 200) {
            log.warn("线程数过多: {}", threadCount);
        }
    }
}

@Component
public class PerformanceMonitor {
    private static final Logger log = LoggerFactory.getLogger(PerformanceMonitor.class);

    @Around("@annotation(Monitored)")
    public Object monitor(ProceedingJoinPoint joinPoint) throws Throwable {
        long startTime = System.currentTimeMillis();
        String methodName = joinPoint.getSignature().getName();

        try {
            Object result = joinPoint.proceed();
            long duration = System.currentTimeMillis() - startTime;

            if (duration > 1000) {
                log.warn("方法 {} 执行时间过长: {}ms", methodName, duration);
            } else {
                log.debug("方法 {} 执行时间: {}ms", methodName, duration);
            }

            return result;
        } catch (Exception e) {
            long duration = System.currentTimeMillis() - startTime;
            log.error("方法 {} 执行失败，耗时: {}ms", methodName, duration, e);
            throw e;
        }
    }
}

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface Monitored {
}

@SpringBootApplication
public class ModelRouterApplication {

    public static void main(String[] args) {
        // 启用 Reactor 调试
        Hooks.onOperatorDebug();

        SpringApplication.run(ModelRouterApplication.class, args);
    }
}

@Service
public class ChatService {
    private static final Logger log = LoggerFactory.getLogger(ChatService.class);

    public Mono<String> processChat(String request) {
        return Mono.fromCallable(() -> request)
            .doOnSubscribe(s -> log.debug("开始处理聊天请求"))
            .doOnNext(req -> log.debug("处理请求: {}", req))
            .flatMap(this::callBackend)
            .doOnSuccess(response -> log.debug("后端响应: {}", response))
            .doOnError(error -> log.error("处理失败", error))
            .doFinally(signal -> log.debug("处理完成: {}", signal));
    }

    private Mono<String> callBackend(String request) {
        return webClient.post()
            .uri("/v1/chat/completions")
            .bodyValue(request)
            .retrieve()
            .bodyToMono(String.class)
            .timeout(Duration.ofSeconds(30))
            .onErrorMap(TimeoutException.class, 
                ex -> new ServiceException("后端服务超时", ex));
    }
}

@Component
public class BackpressureMonitor {
    private static final Logger log = LoggerFactory.getLogger(BackpressureMonitor.class);

    public Flux<String> processStream(Flux<String> input) {
        return input
            .onBackpressureBuffer(1000, 
                dropped -> log.warn("背压缓冲区满，丢弃元素: {}", dropped))
            .doOnRequest(n -> log.debug("请求 {} 个元素", n))
            .doOnCancel(() -> log.debug("流被取消"))
            .map(this::processItem);
    }
}

@ConfigurationProperties(prefix = "model")
@Validated
@Data
public class ModelConfiguration {

    @PostConstruct
    public void validate() {
        log.info("模型配置加载完成: {}", this);

        if (services == null || services.isEmpty()) {
            throw new IllegalStateException("至少需要配置一个服务");
        }

        services.forEach((type, config) -> {
            if (config.getInstances() == null || config.getInstances().isEmpty()) {
                throw new IllegalStateException("服务 " + type + " 至少需要一个实例");
            }
        });
    }
}

@Component
public class ConfigurationChangeListener {
    private static final Logger log = LoggerFactory.getLogger(ConfigurationChangeListener.class);

    @EventListener
    public void handleConfigurationChange(ConfigurationChangeEvent event) {
        log.info("配置变更: 类型={}, 键={}, 旧值={}, 新值={}", 
                event.getType(), event.getKey(), event.getOldValue(), event.getNewValue());
    }
}

@Component
public class BackendHealthIndicator implements HealthIndicator {

    @Override
    public Health health() {
        try {
            // 检查后端服务健康状态
            boolean isHealthy = checkBackendHealth();

            if (isHealthy) {
                return Health.up()
                    .withDetail("backend", "All services are healthy")
                    .build();
            } else {
                return Health.down()
                    .withDetail("backend", "Some services are unhealthy")
                    .build();
            }
        } catch (Exception e) {
            return Health.down(e)
                .withDetail("error", e.getMessage())
                .build();
        }
    }
}

@Component
public class CustomMetrics {
    private final Counter requestCounter;
    private final Timer requestTimer;

    public CustomMetrics(MeterRegistry meterRegistry) {
        this.requestCounter = Counter.builder("jairouter.requests.custom")
            .description("自定义请求计数器")
            .register(meterRegistry);

        this.requestTimer = Timer.builder("jairouter.request.duration.custom")
            .description("自定义请求耗时")
            .register(meterRegistry);
    }

    public void recordRequest(String serviceType, Duration duration) {
        requestCounter.increment(Tags.of("service", serviceType));
        requestTimer.record(duration, Tags.of("service", serviceType));
    }
}