#!/bin/bash
# 虚拟线程生产环境JVM启动脚本

# 基础JVM配置
JAVA_OPTS="-server"

# 内存配置 - 虚拟线程本身内存占用很少，但要为应用数据预留足够空间
JAVA_OPTS="$JAVA_OPTS -Xms4g -Xmx8g"
JAVA_OPTS="$JAVA_OPTS -XX:MaxDirectMemorySize=2g"
JAVA_OPTS="$JAVA_OPTS -XX:MetaspaceSize=256m -XX:MaxMetaspaceSize=512m"

# 垃圾收集器配置 - 推荐使用ZGC以获得更好的延迟表现
JAVA_OPTS="$JAVA_OPTS -XX:+UnlockExperimentalVMOptions"
JAVA_OPTS="$JAVA_OPTS -XX:+UseZGC"
JAVA_OPTS="$JAVA_OPTS -XX:+UnlockDiagnosticVMOptions"

# 虚拟线程特定配置
JAVA_OPTS="$JAVA_OPTS -Djdk.virtualThreadScheduler.parallelism=32"
JAVA_OPTS="$JAVA_OPTS -Djdk.virtualThreadScheduler.maxPoolSize=256"

# 调试和监控配置（生产环境可选）
JAVA_OPTS="$JAVA_OPTS -Djdk.tracePinnedThreads=short"  # 生产环境使用short而不是full

# JFR配置 - 用于性能分析
JAVA_OPTS="$JAVA_OPTS -XX:+FlightRecorder"
JAVA_OPTS="$JAVA_OPTS -XX:StartFlightRecording=duration=300s,filename=app-performance.jfr"

# 其他优化参数
JAVA_OPTS="$JAVA_OPTS -XX:+UseCompressedOops"
JAVA_OPTS="$JAVA_OPTS -XX:+UseCompressedClassPointers"
JAVA_OPTS="$JAVA_OPTS -XX:+OptimizeStringConcat"

# 启动应用
java $JAVA_OPTS -jar your-application.jar

@Component
@Slf4j
public class VirtualThreadMonitoring {

    private final MeterRegistry meterRegistry;
    private final ScheduledExecutorService scheduler;

    public VirtualThreadMonitoring(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
        this.scheduler = Executors.newScheduledThreadPool(1);

        // 启动监控
        startMonitoring();
    }

    /**
     * 启动虚拟线程监控
     */
    private void startMonitoring() {
        // 每30秒收集一次虚拟线程指标
        scheduler.scheduleAtFixedRate(this::collectVirtualThreadMetrics,
            0, 30, TimeUnit.SECONDS);

        // 注册自定义指标
        registerCustomMetrics();
    }

    /**
     * 收集虚拟线程相关指标
     */
    private void collectVirtualThreadMetrics() {
        try {
            var threadMXBean = ManagementFactory.getThreadMXBean();

            // 平台线程数量
            var platformThreadCount = threadMXBean.getThreadCount();
            Gauge.builder("jvm.threads.platform.count")
                .register(meterRegistry, platformThreadCount);

            // 虚拟线程数量（通过Thread.getAllStackTraces()间接获取）
            var allThreads = Thread.getAllStackTraces().keySet();
            var virtualThreadCount = allThreads.stream()
                .mapToInt(thread -> thread.isVirtual() ? 1 : 0)
                .sum();

            Gauge.builder("jvm.threads.virtual.count")
                .register(meterRegistry, virtualThreadCount);

            // 载体线程池信息
            collectCarrierThreadPoolMetrics();

            log.debug("虚拟线程监控指标 - Platform: {}, Virtual: {}",
                platformThreadCount, virtualThreadCount);

        } catch (Exception e) {
            log.error("收集虚拟线程指标失败", e);
        }
    }

    /**
     * 收集载体线程池指标
     */
    private void collectCarrierThreadPoolMetrics() {
        // 这里需要使用反射或JFR API来获取载体线程池信息
        // 实际实现会更复杂，需要访问内部API

        try {
            // 示例：通过系统属性获取调度器信息
            var parallelism = Integer.getInteger("jdk.virtualThreadScheduler.parallelism",
                Runtime.getRuntime().availableProcessors());
            var maxPoolSize = Integer.getInteger("jdk.virtualThreadScheduler.maxPoolSize", 256);

            Gauge.builder("jvm.threads.virtual.scheduler.parallelism")
                .register(meterRegistry, parallelism);

            Gauge.builder("jvm.threads.virtual.scheduler.max_pool_size")
                .register(meterRegistry, maxPoolSize);

        } catch (Exception e) {
            log.warn("无法获取载体线程池指标", e);
        }
    }

    /**
     * 注册自定义业务指标
     */
    private void registerCustomMetrics() {
        // 虚拟线程创建计数器
        Counter.builder("app.virtual.threads.created")
            .description("虚拟线程创建总数")
            .register(meterRegistry);

        // 虚拟线程任务执行时间
        Timer.builder("app.virtual.threads.task.duration")
            .description("虚拟线程任务执行时间")
            .register(meterRegistry);

        // 线程池满时的等待时间
        Timer.builder("app.virtual.threads.queue.wait")
            .description("任务等待执行时间")
            .register(meterRegistry);
    }

    /**
     * 记录虚拟线程任务执行
     */
    public <T> T recordVirtualThreadTask(String taskName, Supplier<T> task) {
        var timer = Timer.builder("app.virtual.threads.task.duration")
            .tag("task", taskName)
            .register(meterRegistry);

        return timer.recordCallable(() -> {
            // 增加创建计数
            meterRegistry.counter("app.virtual.threads.created",
                "task", taskName).increment();

            return task.get();
        });
    }

    @PreDestroy
    public void shutdown() {
        scheduler.shutdown();
        try {
            if (!scheduler.awaitTermination(10, TimeUnit.SECONDS)) {
                scheduler.shutdownNow();
            }
        } catch (InterruptedException e) {
            scheduler.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }
}

@Component
@Slf4j
public class VirtualThreadBestPractices {

    /**
     *  最适合虚拟线程的场景
     * 1. I/O密集型任务
     * 2. 大量并发连接
     * 3. 阻塞式API调用
     */
    public void ioIntensiveExample() {
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {

            // 数千个并发数据库查询
            var tasks = IntStream.range(0, 10000)
                .mapToObj(i -> (Runnable) () -> {
                    try {
                        // 模拟数据库查询
                        var result = databaseService.complexQuery(i);
                        // 模拟外部API调用
                        var enrichedData = externalApiService.enrichData(result);
                        // 处理结果
                        processResult(enrichedData);
                    } catch (Exception e) {
                        log.error("任务 {} 执行失败", i, e);
                    }
                })
                .toList();

            tasks.forEach(executor::execute);
        }
    }

    /**
     *  不适合虚拟线程的场景
     * 1. CPU密集型计算
     * 2. 同步阻塞操作（如synchronized块）
     * 3. 原生方法调用
     */
    public void cpuIntensiveCounterExample() {
        // 对于CPU密集型任务，传统线程池更合适
        var cpuIntensivePool = Executors.newFixedThreadPool(
            Runtime.getRuntime().availableProcessors());

        try {
            var futures = IntStream.range(0, 1000)
                .mapToObj(i -> cpuIntensivePool.submit(() -> {
                    // CPU密集型计算
                    return performComplexCalculation(i);
                }))
                .toList();

            futures.forEach(future -> {
                try {
                    var result = future.get();
                    log.debug("计算结果: {}", result);
                } catch (Exception e) {
                    log.error("计算失败", e);
                }
            });
        } finally {
            cpuIntensivePool.shutdown();
        }
    }

    /**
     *  混合场景：智能选择执行器
     */
    public void hybridTaskExample() {
        var ioExecutor = Executors.newVirtualThreadPerTaskExecutor();
        var cpuExecutor = Executors.newFixedThreadPool(
            Runtime.getRuntime().availableProcessors());

        try {
            var tasks = generateMixedTasks(1000);

            tasks.forEach(task -> {
                if (task.isIOIntensive()) {
                    ioExecutor.execute(() -> processIOTask(task));
                } else {
                    cpuExecutor.execute(() -> processCPUTask(task));
                }
            });

        } finally {
            ioExecutor.close();
            cpuExecutor.shutdown();
        }
    }

    /**
     *  性能基准测试框架
     */
    public PerformanceTestResult runPerformanceBenchmark(BenchmarkConfig config) {
        log.info("开始性能基准测试 - 配置: {}", config);

        var results = new ArrayList<TestResult>();

        // 测试传统线程池
        var traditionalResult = benchmarkTraditionalThreads(config);
        results.add(traditionalResult);

        // 测试虚拟线程
        var virtualResult = benchmarkVirtualThreads(config);
        results.add(virtualResult);

        // 分析结果
        var analysis = analyzeResults(traditionalResult, virtualResult);

        return PerformanceTestResult.builder()
            .config(config)
            .results(results)
            .analysis(analysis)
            .timestamp(Instant.now())
            .build();
    }

    private TestResult benchmarkTraditionalThreads(BenchmarkConfig config) {
        var startTime = System.nanoTime();

        try (var executor = Executors.newFixedThreadPool(config.threadPoolSize())) {
            var futures = IntStream.range(0, config.taskCount())
                .mapToObj(i -> executor.submit(() -> simulateTask(config.taskDuration())))
                .toList();

            futures.forEach(future -> {
                try {
                    future.get();
                } catch (Exception e) {
                    log.error("传统线程任务执行失败", e);
                }
            });
        }

        var duration = Duration.ofNanos(System.nanoTime() - startTime);

        return TestResult.builder()
            .executorType("Traditional ThreadPool")
            .taskCount(config.taskCount())
            .totalDuration(duration)
            .throughput((double) config.taskCount() / duration.toSeconds())
            .averageTaskTime(duration.dividedBy(config.taskCount()))
            .build();
    }

    private TestResult benchmarkVirtualThreads(BenchmarkConfig config) {
        var startTime = System.nanoTime();

        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            var futures = IntStream.range(0, config.taskCount())
                .mapToObj(i -> executor.submit(() -> simulateTask(config.taskDuration())))
                .toList();

            futures.forEach(future -> {
                try {
                    future.get();
                } catch (Exception e) {
                    log.error("虚拟线程任务执行失败", e);
                }
            });
        }

        var duration = Duration.ofNanos(System.nanoTime() - startTime);

        return TestResult.builder()
            .executorType("Virtual Threads")
            .taskCount(config.taskCount())
            .totalDuration(duration)
            .throughput((double) config.taskCount() / duration.toSeconds())
            .averageTaskTime(duration.dividedBy(config.taskCount()))
            .build();
    }

    private void simulateTask(Duration duration) {
        try {
            Thread.sleep(duration);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    private PerformanceAnalysis analyzeResults(TestResult traditional, TestResult virtual) {
        var throughputImprovement = (virtual.throughput() - traditional.throughput())
            / traditional.throughput() * 100;

        var responseTimeImprovement = (traditional.averageTaskTime().toNanos()
            - virtual.averageTaskTime().toNanos())
            / (double) traditional.averageTaskTime().toNanos() * 100;

        return PerformanceAnalysis.builder()
            .throughputImprovement(throughputImprovement)
            .responseTimeImprovement(responseTimeImprovement)
            .recommendation(generateRecommendation(throughputImprovement, responseTimeImprovement))
            .build();
    }

    private String generateRecommendation(double throughputImprovement, double responseTimeImprovement) {
        if (throughputImprovement > 100 && responseTimeImprovement > 50) {
            return "强烈推荐使用虚拟线程，性能提升显著";
        } else if (throughputImprovement > 50) {
            return "推荐使用虚拟线程，有明显性能提升";
        } else if (throughputImprovement > 0) {
            return "可以考虑使用虚拟线程，有一定性能提升";
        } else {
            return "当前场景下传统线程池可能更合适";
        }
    }

    // 数据类定义
    public record BenchmarkConfig(
        int taskCount,
        Duration taskDuration,
        int threadPoolSize
    ) {}

    @Builder
    public record TestResult(
        String executorType,
        int taskCount,
        Duration totalDuration,
        double throughput,
        Duration averageTaskTime
    ) {}

    @Builder
    public record PerformanceAnalysis(
        double throughputImprovement,
        double responseTimeImprovement,
        String recommendation
    ) {}

    @Builder
    public record PerformanceTestResult(
        BenchmarkConfig config,
        List<TestResult> results,
        PerformanceAnalysis analysis,
        Instant timestamp
    ) {}
}

@Component
@Slf4j
public class ProductionMetricsCollector {

    private final MeterRegistry meterRegistry;
    private final ApplicationEventPublisher eventPublisher;

    /**
     * 生产环境关键指标收集
     */
    @Scheduled(fixedRate = 60000) // 每分钟收集一次
    public void collectProductionMetrics() {
        try {
            var metrics = ProductionMetrics.builder()
                .timestamp(Instant.now())
                .virtualThreadCount(countActiveVirtualThreads())
                .carrierThreadCount(countCarrierThreads())
                .pinnedThreadCount(countPinnedThreads())
                .averageTaskDuration(calculateAverageTaskDuration())
                .throughput(calculateCurrentThroughput())
                .memoryUsage(getCurrentMemoryUsage())
                .cpuUsage(getCurrentCpuUsage())
                .build();

            // 记录指标
            recordMetrics(metrics);

            // 检查异常情况
            checkForAnomalies(metrics);

            log.debug("生产环境指标收集完成: {}", metrics);

        } catch (Exception e) {
            log.error("生产环境指标收集失败", e);
        }
    }

    /**
     * 异常检测和告警
     */
    private void checkForAnomalies(ProductionMetrics metrics) {
        var anomalies = new ArrayList<String>();

        // 检查虚拟线程数量异常
        if (metrics.virtualThreadCount() > 1_000_000) {
            anomalies.add("虚拟线程数量过多: " + metrics.virtualThreadCount());
        }

        // 检查线程固定问题
        if (metrics.pinnedThreadCount() > metrics.carrierThreadCount() * 0.8) {
            anomalies.add("大量线程被固定: " + metrics.pinnedThreadCount() + "/" + metrics.carrierThreadCount());
        }

        // 检查内存使用异常
        if (metrics.memoryUsage() > 0.9) {
            anomalies.add("内存使用率过高: " + String.format("%.1f%%", metrics.memoryUsage() * 100));
        }

        // 检查吞吐量下降
        var previousThroughput = getPreviousThroughput();
        if (previousThroughput > 0 && metrics.throughput() < previousThroughput * 0.7) {
            anomalies.add("吞吐量显著下降: " + metrics.throughput() + " (之前: " + previousThroughput + ")");
        }

        // 发送告警
        if (!anomalies.isEmpty()) {
            var alert = VirtualThreadAlert.builder()
                .timestamp(Instant.now())
                .severity(AlertSeverity.WARNING)
                .anomalies(anomalies)
                .metrics(metrics)
                .build();

            eventPublisher.publishEvent(alert);
            log.warn("检测到虚拟线程异常: {}", anomalies);
        }
    }

    /**
     * 生成性能报告
     */
    @Scheduled(cron = "0 0 1 * * ?") // 每天凌晨1点生成报告
    public void generateDailyReport() {
        try {
            var report = createDailyPerformanceReport();
            saveReportToFile(report);

            log.info("每日性能报告已生成: {}", report.getReportFile());

        } catch (Exception e) {
            log.error("生成每日报告失败", e);
        }
    }

    private DailyPerformanceReport createDailyPerformanceReport() {
        var yesterday = LocalDate.now().minusDays(1);
        var metrics = getMetricsForDate(yesterday);

        return DailyPerformanceReport.builder()
            .date(yesterday)
            .totalRequests(metrics.stream().mapToLong(m -> (long) m.throughput()).sum())
            .averageResponseTime(calculateAverageResponseTime(metrics))
            .peakConcurrency(metrics.stream().mapToInt(ProductionMetrics::virtualThreadCount).max().orElse(0))
            .errorRate(calculateErrorRate(metrics))
            .memoryUsageStats(calculateMemoryStats(metrics))
            .recommendations(generateDailyRecommendations(metrics))
            .build();
    }

    // 辅助方法和数据类
    @Builder
    public record ProductionMetrics(
        Instant timestamp,
        int virtualThreadCount,
        int carrierThreadCount,
        int pinnedThreadCount,
        Duration averageTaskDuration,
        double throughput,
        double memoryUsage,
        double cpuUsage
    ) {}

    @Builder
    public record VirtualThreadAlert(
        Instant timestamp,
        AlertSeverity severity,
        List<String> anomalies,
        ProductionMetrics metrics
    ) {}

    public enum AlertSeverity {
        INFO, WARNING, ERROR, CRITICAL
    }

    @Builder
    public record DailyPerformanceReport(
        LocalDate date,
        long totalRequests,
        Duration averageResponseTime,
        int peakConcurrency,
        double errorRate,
        MemoryUsageStats memoryUsageStats,
        List<String> recommendations
    ) {
        public String getReportFile() {
            return "virtual-thread-report-" + date + ".json";
        }
    }

    @Builder
    public record MemoryUsageStats(
        double average,
        double peak,
        double minimum
    ) {}
}