03_并发执行

发表于 更新于 分类于 ， Changyan： 本文字数： 4k 阅读时长 ≈ 14 分钟

掌握了async/await基础后，本课教你如何让多个异步任务真正”同时”运行。我们将学习三个核心工具：gather（等所有任务完成）、create_task（创建后台任务）、wait（灵活控制）。就像餐厅有多个服务员同时服务不同桌客人。通过批量下载、限流控制、超时管理等实战案例，你将能够编写高效的并发程序，实现10-100倍的性能提升。

📖 课程目标

  • 掌握 asyncio.gather() 的使用
  • 理解 asyncio.create_task() 的作用
  • 学会使用 asyncio.wait() 控制任务
  • 理解并发 vs 并行的区别
  • 编写高效的并发程序

🎯 什么是并发？

生活类比：餐厅服务员

场景1：只有1个服务员（并发）

1
2
3
4
5
6
7
8
9
10
11
服务员：
  去1号桌点餐 → 记下订单 → 
  去2号桌点餐 → 记下订单 →
  去3号桌点餐 → 记下订单 →
  回厨房下单 →
  去4号桌点餐 → ...

特点：
  - 1个人处理多个任务
  - 快速切换，不让客人等太久
  - 这就是"并发"！

场景2：有3个服务员（并行）

1
2
3
4
5
6
7
8
服务员A：专门负责1-3号桌
服务员B：专门负责4-6号桌
服务员C：专门负责7-9号桌

特点：
  - 多个人同时工作
  - 真正的"同时"
  - 这就是"并行"！

编程中的并发 vs 并行

概念说明实现方式适用场景
并发快速切换任务，看起来像同时asyncio（单线程）I/O密集型
并行真正同时执行多个任务multiprocessing（多进程）CPU密集型

本课重点：并发（asyncio）

🔧 方法1：asyncio.gather()

基本用法

gather() 用于同时运行多个协程，等待全部完成。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import asyncio

async def task1():
    await asyncio.sleep(2)
    return "任务1完成"

async def task2():
    await asyncio.sleep(3)
    return "任务2完成"

async def main():
    # 同时运行多个任务
    result = await asyncio.gather(
        task1(),
        task2()
    )
    print(result)  # ['任务1完成', '任务2完成']

asyncio.run(main())

特点

  • ✅ 等待所有任务完成

  • ✅ 按顺序返回结果（列表）

  • ✅ 如果某个任务出错，会抛出异常

  • ✅ 可以用 return_exceptions=True 捕获异常

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    import asyncio
    import time


    async def task1():
        await asyncio.sleep(3)
        raise Exception("task1 异常333")
        return "task1 completed"

    async def task2():
        await asyncio.sleep(2)
        return "task2 completed"

    async def main():
        result = await asyncio.gather(task1(), task2(), return_exceptions=True)
        print(result)
    start = time.time()
    asyncio.run(main())
    print('duration：', time.time() - start)

    输出：

    1
    2
    [Exception('task1 异常333'), 'task2 完成']
    耗时： 3.002075910568237

生活类比：团队协作

1
2
3
4
5
6
7
8
老板：小明、小红、小李，你们同时去完成各自的任务！
      我等你们全部完成后，再开会总结。

小明：[做任务A] → 完成
小红：[做任务B] → 完成
小李：[做任务C] → 完成

老板：好的，大家都完成了，我们开会！

🔧 方法2：asyncio.create_task()

基本用法

create_task() 用于创建一个后台任务，立即开始执行。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import asyncio

async def background_task():
    print("后台任务开始")
    await asyncio.sleep(3)
    print("后台任务完成")
    return "result"

async def main():
    # 创建任务（立即开始执行）
    task = asyncio.create_task(background_task())
    
    # 做其他事情
    print("主程序继续执行...")
    await asyncio.sleep(1)
    print("主程序做了一些事情")
    
    # 等待任务完成
    result = await task
    print(f"任务结果：{result}")

asyncio.run(main())

特点

  • ✅ 创建后立即开始执行
  • ✅ 不阻塞当前代码
  • ✅ 可以稍后等待结果
  • ✅ 适合”发起任务后继续做其他事”的场景

生活类比：洗衣服

1
2
3
4
5
6
7
8
你：把衣服放进洗衣机，按下启动（create_task）
洗衣机：[开始洗衣服...]

你：去做其他事情（看书、做饭）
洗衣机：[继续洗...]

你：过了一会儿，去看洗衣机（await 任务）
洗衣机：洗好了！

🔧 方法3：asyncio.wait()

基本用法

wait() 提供更灵活的任务控制。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import asyncio

async def main():
    task_list = [
        asyncio.create_task(task1()),
        asyncio.create_task(task2()),
        asyncio.create_task(task3())
    ]
    
    # 等待所有任务完成
    completed, not_completed = await asyncio.wait(task_list)
    
    # 处理结果
    for task in completed:
        result = await task
        print(result)

等待策略

策略说明使用场景
FIRST_COMPLETED任意一个完成就返回竞速场景
FIRST_EXCEPTION第一个异常就返回快速失败
ALL_COMPLETED全部完成才返回（默认）等待全部

生活类比：比赛

1
2
3
4
5
6
7
8
场景1：ALL_COMPLETED（等全部完成）
  裁判：等所有选手都跑完，再公布成绩

场景2：FIRST_COMPLETED（任意一个完成）
  裁判：第一个冲线的就是冠军，比赛结束！

场景3：FIRST_EXCEPTION（第一个出错）
  裁判：有人犯规了，比赛暂停！

📊 三种方法对比

方法使用场景优点缺点
gather()同时运行多个任务，需要所有结果简单易用不够灵活
create_task()创建后台任务，稍后等待灵活控制需要手动管理
wait()需要灵活控制等待策略功能强大稍微复杂

选择建议

1
2
3
4
5
6
7
8
9
10
11
# 场景1：同时运行，等待所有结果
result = await asyncio.gather(task1(), task2(), task3())

# 场景2：创建任务，稍后等待
task = asyncio.create_task(background_task())
# ... 做其他事情 ...
result = await task

# 场景3：需要灵活控制
task_list = [asyncio.create_task(t) for t in tasks]
completed, not_completed = await asyncio.wait(task_list, return_when=FIRST_COMPLETED)

💡 实战案例：批量下载器

需求

实现一个批量下载器，可以：

  1. 同时下载多个文件
  2. 显示下载进度
  3. 统计总耗时

代码示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
import asyncio
import time

async def download_file(filename: str, size_mb: int) -> dict:
    """模拟下载文件"""
    print(f"📥 开始下载：{filename} ({size}MB)")
    
    start_time = time.time()
    
    # 模拟下载（每MB需要0.5秒）
    download_time = size * 0.5
    await asyncio.sleep(download_time)
    
    duration = time.time() - start_time
    
    print(f"✅ 下载完成：{filename} (duration{duration:.1f}秒)")
    
    return {
        "filename": filename,
        "size": size,
        "duration": duration
    }

async def batch_downloader():
    """批量下载多个文件"""
    file_list = [
        ("Python教程.pdf", 10),
        ("数据分析.xlsx", 5),
        ("项目代码.zip", 15),
        ("演示视频.mp4", 20),
        ("文档.docx", 3)
    ]
    
    print("🚀 开始批量下载...")
    total_start_time = time.time()
    
    # 方法1：使用 gather
    task_list = [download_file(name, size) for name, size in file_list]
    result_list = await asyncio.gather(*task_list)
    
    total_time = time.time() - total_start_time
    
    # 统计
    print("\n" + "=" * 50)
    print("📊 下载统计")
    print("=" * 50)
    total_size = sum(r["size"] for r in result_list)
    print(f"文件数量：{len(result_list)}个")
    print(f"total_size：{total_size}MB")
    print(f"total_time：{总duration:.1f}秒")
    print(f"平均速度：{total_size/总duration:.1f}MB/秒")

完整代码在 03_examples.py 中！

🎯 性能对比

场景：下载5个文件

文件大小下载时间
文件110MB5秒
文件25MB2.5秒
文件315MB7.5秒
文件420MB10秒
文件53MB1.5秒

结果对比

方式总耗时效率
同步（一个一个下载）26.5秒基准
异步（同时下载）10秒2.65倍 ⚡

🎪 高级技巧

技巧1：限制并发数量

有时候不能同时运行太多任务（比如API有限流）。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import asyncio

async def limited_task(number: int, semaphore: asyncio.Semaphore):
    """使用信号量限制并发数"""
    async with semaphore:  # 获取许可
        print(f"task{number}开始")
        await asyncio.sleep(2)
        print(f"task{number}completed")

async def main():
    # 最多同时运行3个任务
    semaphore = asyncio.Semaphore(3)
    
    task_list = [limited_task(i, semaphore) for i in range(10)]
    await asyncio.gather(*task_list)

技巧2：超时控制

1
2
3
4
5
6
7
8
9
10
11
12
import asyncio

async def potentially_slow_task():
    await asyncio.sleep(10)
    return "completed"

async def main():
    try:
        # 最多等待5秒
        result = await asyncio.wait_for(potentially_slow_task(), timeout=5)
    except asyncio.TimeoutError:
        print("任务超时！")

技巧3：任务取消

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import asyncio

async def long_task():
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        print("任务被取消了")
        raise

async def main():
    task = asyncio.create_task(long_task())
    
    await asyncio.sleep(2)
    
    # 取消任务
    task.cancel()
    
    try:
        await task
    except asyncio.CancelledError:
        print("确认任务已取消")

📝 本课小结

核心方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 1. gather - 同时运行，等待全部
result = await asyncio.gather(task1(), task2(), task3())

# 2. create_task - 创建后台任务
task = asyncio.create_task(async_function())
result = await task

# 3. wait - 灵活控制
completed, not_completed = await asyncio.wait(task_list)

# 4. Semaphore - 限制并发数
semaphore = asyncio.Semaphore(3)
async with semaphore:
    # 受限的操作

# 5. wait_for - 超时控制
result = await asyncio.wait_for(task(), timeout=5)

使用场景

  • gather：批量下载、批量API调用
  • create_task：后台任务、定时任务
  • wait：竞速场景、需要部分结果
  • Semaphore：API限流、资源限制
  • wait_for：防止任务卡死

🎯 下一步

  1. 运行 03_examples.py 查看完整示例
  2. 尝试修改并发数量，观察效果
  3. 完成课后练习题
  4. 准备学习第4课：异步网络请求实战

💪 课后练习

练习1：批量查询

编写一个程序，同时查询多个城市的天气（模拟）。

1
2
3
4
5
6
7
8
9
async def query_weather(city: str) -> dict:
    # 你的代码
    pass

async def main():
    city_list = ["北京", "上海", "广州", "深圳", "杭州"]
    # 同时查询所有城市
    # 你的代码
    pass

练习2：限流下载

实现一个下载器，最多同时下载3个文件。

1
2
3
async def download_file(filename: str, semaphore: asyncio.Semaphore):
    # 你的代码
    pass

练习3：超时重试

实现一个带超时和重试的任务执行器。

1
2
3
async def execute_task(task_name: str, timeout_seconds: int, max_retry: int):
    # 你的代码
    pass

答案在 练习答案.py 中！ 😊

03_examples.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
"""
第3课示例代码：并发执行多个任务

运行方式：
    python 03_examples.py
"""

import asyncio
import time
from typing import List, Dict, Set


# ============================================
# 示例1：asyncio.gather() 基础用法
# ============================================

async def task_A() -> str:
    """任务A：需要2秒"""
    print("🅰️  任务A开始")
    await asyncio.sleep(2)
    print("🅰️  任务A完成")
    return "任务A的结果"


async def task_B() -> str:
    """任务B：需要3秒"""
    print("🅱️  任务B开始")
    await asyncio.sleep(3)
    print("🅱️  任务B完成")
    return "任务B的结果"


async def task_C() -> str:
    """任务C：需要1秒"""
    print("🅲  任务C开始")
    await asyncio.sleep(1)
    print("🅲  任务C完成")
    return "任务C的结果"


async def example1_gather_basics() -> None:
    """示例1：使用 gather 同时运行多个任务"""
    print("\n" + "=" * 50)
    print("📚 示例1：asyncio.gather() 基础用法")
    print("=" * 50)

    start_time = time.time()

    # 同时运行三个任务
    result_list = await asyncio.gather(
        task_A(),
        task_B(),
        task_C()
    )

    total_time = time.time() - start_time

    print(f"\n✅ 所有任务完成！")
    print(f"result_list：{result_list}")
    print(f"⏱️  total_time：{总duration:.1f}秒（最长的任务是3秒）")

    print("\n💡 关键点：")
    print("  1. gather 会等待所有任务完成")
    print("  2. 返回结果是列表，顺序与传入顺序一致")
    print("  3. total_time = 最长任务的时间（而不是总和）")


# ============================================
# 示例2：asyncio.create_task() 用法
# ============================================

async def background_task(task_name: str, duration: int) -> str:
    """一个后台任务"""
    print(f"🔧 {任务名}开始执行...")
    await asyncio.sleep(duration)
    print(f"✅ {任务名}执行完成！")
    return f"{任务名}的结果"


async def example2_create_task() -> None:
    """示例2：使用 create_task 创建后台任务"""
    print("\n" + "=" * 50)
    print("📚 示例2：asyncio.create_task() 用法")
    print("=" * 50)

    print("主程序：开始执行")

    # 创建任务（立即开始执行，不等待）
    task1 = asyncio.create_task(background_task("后台任务1", 3))
    task2 = asyncio.create_task(background_task("后台任务2", 2))

    print("主程序：任务已创建，继续做其他事情...")

    # 模拟主程序做其他事情
    for i in range(3):
        await asyncio.sleep(1)
        print(f"主程序：正在做第{i+1}件事...")

    print("主程序：现在等待后台任务完成...")

    # 等待任务完成
    result1 = await task1
    result2 = await task2

    print(f"\n✅ 所有任务完成！")
    print(f"result1：{result1}")
    print(f"result2：{result2}")

    print("\n💡 关键点：")
    print("  1. create_task 创建后立即开始执行")
    print("  2. 不会阻塞当前代码")
    print("  3. 可以稍后用 await 等待结果")


# ============================================
# 示例3：asyncio.wait() 灵活控制
# ============================================

async def racing_task(participant: str, speed: int) -> str:
    """竞速任务"""
    print(f"🏃 {选手}开始跑步...")
    await asyncio.sleep(speed)
    print(f"🏁 {选手}到达终点！")
    return f"{选手}completed"


async def example3_wait_control() -> None:
    """示例3：使用 wait 灵活控制任务"""
    print("\n" + "=" * 50)
    print("📚 示例3：asyncio.wait() 灵活控制")
    print("=" * 50)

    # 创建多个任务
    task_list = [
        asyncio.create_task(racing_task("选手A", 3)),
        asyncio.create_task(racing_task("选手B", 2)),
        asyncio.create_task(racing_task("选手C", 4)),
    ]

    print("🏁 比赛开始！等待第一个完成...")

    # 等待第一个完成
    completed, not_completed = await asyncio.wait(
        task_list,
        return_when=asyncio.FIRST_COMPLETED
    )

    print(f"\n🎉 第一名产生了！")
    for task in completed:
        result = await task
        print(f"冠军：{result}")

    print(f"\n还有 {len(not_completed)} 个选手在跑...")

    # 等待剩余任务完成
    if not_completed:
        await asyncio.wait(not_completed)
        print("所有选手都完成了！")

    print("\n💡 关键点：")
    print("  1. wait 可以灵活控制等待策略")
    print("  2. FIRST_COMPLETED：第一个完成就返回")
    print("  3. 返回两个集合：完成的和未完成的")


# ============================================
# 示例4：实战 - 批量下载器
# ============================================

async def download_file(filename: str, size_mb: int) -> Dict[str, any]:
    """模拟下载文件"""
    print(f"📥 开始下载：{filename} ({size}MB)")

    start_time = time.time()

    # 模拟下载（每MB需要0.3秒）
    download_time = size * 0.3

    # 显示进度
    step_count = 5
    for step in range(1, step_count + 1):
        await asyncio.sleep(download_time / step_count)
        progress = (step / step_count) * 100
        print(f"   {filename}: {progress:.0f}%")

    duration = time.time() - start_time

    print(f"✅ 下载完成：{filename} (duration{duration:.1f}秒)")

    return {
        "filename": filename,
        "size": size,
        "duration": duration
    }


async def batch_download_sync() -> None:
    """同步方式：一个一个下载"""
    print("\n" + "=" * 50)
    print("📚 示例4A：批量下载 - 同步方式")
    print("=" * 50)

    file_list = [
        ("Python教程.pdf", 5),
        ("数据分析.xlsx", 3),
        ("项目代码.zip", 7),
        ("演示视频.mp4", 10),
    ]

    total_start_time = time.time()
    result_list = []

    # 一个一个下载
    for filename, size in file_list:
        result = await download_file(filename, size)
        result_list.append(result)

    total_time = time.time() - total_start_time

    # 统计
    print("\n📊 下载统计")
    total_size = sum(r["size"] for r in result_list)
    print(f"文件数量：{len(result_list)}个")
    print(f"total_size：{total_size}MB")
    print(f"total_time：{总duration:.1f}秒")


async def batch_download_async() -> None:
    """并发方式：同时下载"""
    print("\n" + "=" * 50)
    print("📚 示例4B：批量下载 - 并发方式")
    print("=" * 50)

    file_list = [
        ("Python教程.pdf", 5),
        ("数据分析.xlsx", 3),
        ("项目代码.zip", 7),
        ("演示视频.mp4", 10),
    ]

    total_start_time = time.time()

    # 同时下载所有文件
    task_list = [download_file(filename, size) for filename, size in file_list]
    result_list = await asyncio.gather(*task_list)

    total_time = time.time() - total_start_time

    # 统计
    print("\n📊 下载统计")
    total_size = sum(r["size"] for r in result_list)
    print(f"文件数量：{len(result_list)}个")
    print(f"total_size：{total_size}MB")
    print(f"total_time：{总duration:.1f}秒")
    print(f"💡 效率提升：{7.5/总duration:.1f}倍！")


# ============================================
# 示例5：限制并发数量（Semaphore）
# ============================================

async def limited_download(filename: str, size: int, semaphore: asyncio.Semaphore) -> str:
    """使用信号量限制并发的下载"""
    async with semaphore:  # 获取许可（最多N个同时执行）
        print(f"📥 开始下载：{filename}")
        await asyncio.sleep(size * 0.5)
        print(f"✅ 完成下载：{filename}")
        return filename


async def example5_limit_concurrency() -> None:
    """示例5：使用 Semaphore 限制并发数量"""
    print("\n" + "=" * 50)
    print("📚 示例5：限制并发数量（Semaphore）")
    print("=" * 50)

    # 创建信号量：最多同时2个任务
    semaphore = asyncio.Semaphore(2)

    file_list = [
        ("文件1", 2),
        ("文件2", 2),
        ("文件3", 2),
        ("文件4", 2),
        ("文件5", 2),
    ]

    print("🚀 开始下载（最多同时2个）...")
    start_time = time.time()

    task_list = [limited_download(name, size, semaphore) for name, size in file_list]
    result = await asyncio.gather(*task_list)

    total_time = time.time() - start_time

    print(f"\n✅ 全部完成！total_time：{总duration:.1f}秒")
    print(f"💡 观察：每次最多2个文件在下载")


# ============================================
# 示例6：超时控制
# ============================================

async def potentially_slow_task(task_name: str, duration: int) -> str:
    """一个可能很慢的任务"""
    print(f"🐌 {任务名}开始（需要{duration}秒）...")
    await asyncio.sleep(duration)
    print(f"✅ {任务名}completed！")
    return f"{任务名}的结果"


async def example6_timeout_control() -> None:
    """示例6：使用 wait_for 控制超时"""
    print("\n" + "=" * 50)
    print("📚 示例6：超时控制（wait_for）")
    print("=" * 50)

    # task1：会超时
    print("\n测试1：任务会超时")
    try:
        result = await asyncio.wait_for(
            potentially_slow_task("慢任务", 5),
            timeout=2  # 最多等2秒
        )
        print(f"result：{result}")
    except asyncio.TimeoutError:
        print("❌ 任务超时了！")

    # task2：不会超时
    print("\n测试2：任务不会超时")
    try:
        result = await asyncio.wait_for(
            potentially_slow_task("快任务", 1),
            timeout=3  # 最多等3秒
        )
        print(f"result：{result}")
    except asyncio.TimeoutError:
        print("❌ 任务超时了！")

    print("\n💡 关键点：")
    print("  1. wait_for 可以设置超时时间")
    print("  2. 超时会抛出 TimeoutError 异常")
    print("  3. 可以用 try-except 捕获")


# ============================================
# 示例7：任务取消
# ============================================

async def long_running_task() -> None:
    """一个长时间运行的任务"""
    try:
        print("🔧 长任务开始...")
        for i in range(10):
            await asyncio.sleep(1)
            print(f"   progress：{(i+1)*10}%")
        print("✅ 长任务完成！")
    except asyncio.CancelledError:
        print("❌ 任务被取消了！")
        raise  # 重新抛出，让任务真正取消


async def example7_task_cancellation() -> None:
    """示例7：取消任务"""
    print("\n" + "=" * 50)
    print("📚 示例7：任务取消")
    print("=" * 50)

    # 创建任务
    task = asyncio.create_task(long_running_task())

    # 等待3秒后取消
    await asyncio.sleep(3)

    print("\n⚠️  主程序决定取消任务...")
    task.cancel()

    # 等待任务真正取消
    try:
        await task
    except asyncio.CancelledError:
        print("✅ 确认任务已取消")

    print("\n💡 关键点：")
    print("  1. 使用 task.cancel() 取消任务")
    print("  2. 任务会收到 CancelledError 异常")
    print("  3. 可以在任务中捕获并清理资源")


# ============================================
# 主程序
# ============================================

async def main() -> None:
    """主程序：运行所有示例"""
    print("🎓 第3课：并发执行多个任务")
    print("=" * 50)

    # 运行所有示例
    await example1_gather_basics()
    await example2_create_task()
    await example3_wait_control()
    await batch_download_sync()
    await batch_download_async()
    await example5_limit_concurrency()
    await example6_timeout_control()
    await example7_task_cancellation()

    # 总结
    print("\n" + "=" * 50)
    print("🎉 第3课完成！")
    print("=" * 50)
    print("""
📚 你学到了什么？
    1. asyncio.gather() - 同时运行多个任务
    2. asyncio.create_task() - 创建后台任务
    3. asyncio.wait() - 灵活控制等待策略
    4. asyncio.Semaphore() - 限制并发数量
    5. asyncio.wait_for() - 超时控制
    6. task.cancel() - 取消任务
    
🎯 使用场景：
    - gather：批量下载、批量API调用
    - create_task：后台任务、不阻塞主流程
    - wait：竞速场景、部分完成即可
    - Semaphore：API限流、资源限制
    - wait_for：防止任务卡死
    - cancel：用户取消、超时处理
    
💪 动手练习：
    1. 修改并发数量，观察效果
    2. 尝试不同的等待策略
    3. 实现带重试的下载器
    4. 完成课后练习题
    
🎯 下一步：
    学习异步网络请求实战（第4课）
    """)


if __name__ == "__main__":
    # 运行主程序
    asyncio.run(main())