夢の終着駅

1
@Slf4j
2
public class VirtualThreadDownloader {
3

4
    // 全局信号量控制并发量 (最大50个并发下载)
5
    private static final Semaphore downloadSemaphore = new Semaphore(50);
6

7
    // 原子类保证线程安全
8
    // 正在执行的任务
9
    private static final AtomicInteger activeTasks = new AtomicInteger(0);
10
    // 执行成功的任务
11
    private static final AtomicInteger completedTasks = new AtomicInteger(0);
12
    // 执行失败的任务
13
    private static final AtomicInteger failedTasks = new AtomicInteger(0);
14

15
    public static void main(String[] args) {
16
        // 要下载的URL列表
17
        List<String> urls = List.of(
18
            "https://example.com/file1",
19
            "https://example.com/file2",
20
            // ...
21
            "https://example.com/file100"
22
        );
23

24
        // 启动监控线程
25
        startMonitoringThread();
26

27
        try {
28
            // 执行下载逻辑
29
            downloadAll(urls);
30
            log.info("所有下载任务完成: 成功={}, 失败={}", completedTasks.get(), failedTasks.get());
31
        } catch (Exception e) {
32
            log.error("下载失败！", e);
33
        }
34
    }
35

36
    // 启动监控线程定期报告状态
37
    private static void startMonitoringThread() {
38
        // 使用虚拟线程，不占用资源
39
        Thread monitorThread = Thread.ofVirtual().name("monitor-thread").start(() -> {
40
            while (true) {
41
                try {
42
                    // 每5秒报告一次
43
                    Thread.sleep(5000);
44
                    int active = activeTasks.get();
45
                    int completed = completedTasks.get();
46
                    int failed = failedTasks.get();
47

48
                    // 报告
49
                    log.info("监控报告: 活跃任务={}, 已完成={}, 失败={}, 信号量可用={}",
50
                        active, completed, failed, downloadSemaphore.availablePermits());
51

52
                    // 当所有任务完成时退出监控
53
                    if (active == 0 && (completed + failed) > 0) {
54
                        break;
55
                    }
56
                } catch (InterruptedException e) {
57
                    log.error("监控线程被中断", e);
58
                    // 重置中断标记
59
                    Thread.currentThread().interrupt();
60
                    break;
61
                }
62
            }
63
        });
64
    }
65

66
    // 使用StructuredTaskScope
67
    private static void downloadAll(List<String> urls) throws Exception {
68
        // 创建作用域，使用ShutdownOnFailure策略保障任务必须全部成功，不然可能下载数据不完整导致损坏
69
        try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
70
            // 提交所有下载任务
71
            for (String url : urls) {
72
                scope.fork(() -> downloadUrl(url));
73
            }
74

75
            log.info("已提交所有下载任务，总数: {}", urls.size());
76
            // 等待所有任务完成
77
            scope.join();
78

79
            // 若存在失败结果则抛出异常，中断下载
80
            scope.throwIfFailed();
81
        }
82
    }
83

84
    // 单一职责的下载任务
85
    private static Void downloadUrl(String url) throws Exception {
86
        // cas给执行中任务集合++
87
        activeTasks.incrementAndGet();
88
        log.debug("开始下载: {}", url);
89

90
        try {
91
            // 获取信号量许可
92
            downloadSemaphore.acquire();
93

94
            // 执行下载
95
            fetchUrlContent(url);
96

97
            // 下载完成
98
            completedTasks.incrementAndGet();
99
            log.debug("下载完成: {}", url);
100
            return null;
101

102
        } catch (InterruptedException e) {
103
            log.error("下载被中断: {}", url, e);
104
            failedTasks.incrementAndGet();
105
            Thread.currentThread().interrupt();
106
            throw e;
107
        } catch (Exception e) {
108
            log.error("下载失败: {} - {}", url, e.getMessage());
109
            failedTasks.incrementAndGet();
110
            throw e;
111
        } finally {
112
            activeTasks.decrementAndGet();
113
            // 释放信号量
114
            downloadSemaphore.release();
115
        }
116
    }
117

118
    // 模拟下载
119
    private static voidfetchUrlContent(String url) throws Exception {
120
        // 下载中……
121
    }
122
}