08_Python异步编程练习题

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

本课程包含6节课，从基础到进阶，全面覆盖Python异步编程知识。

祝你在异步编程的道路上越走越远！

使用说明

  1. 每课后都有配套练习题
  2. 先自己尝试完成，再查看答案
  3. 答案在 练习答案.py 文件中
  4. 建议按顺序完成，循序渐进

第1课练习：什么是异步编程

练习1.1：理解概念 ⭐

问题：用自己的话解释什么是异步编程？

提示：想想做早餐的例子

练习1.2：识别场景 ⭐

问题：下面哪些场景适合使用异步编程？

A. 批量下载100个文件
B. 计算圆周率到1000万位
C. 同时查询10个不同的API
D. 对图片进行复杂的滤镜处理
E. 批量读取日志文件

练习1.3：性能计算 ⭐⭐

问题：如果每个网络请求需要0.5秒，同步方式下载50个文件需要多长时间？异步方式大约需要多长时间？

第2课练习：async/await 基础

练习2.1：异步倒计时 ⭐⭐

要求：编写一个异步倒计时函数，从指定秒数倒数到0。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
async def 倒计时(秒数: int) -> None:
    """
    从指定秒数倒数到0
    
    示例：
        await 倒计时(5)
        输出：
        5...
        4...
        3...
        2...
        1...
        时间到！
    """
    # 你的代码
    pass

练习2.2：异步问候 ⭐⭐

要求：编写一个异步函数，向多个人依次问候，每次问候间隔1秒。

1
2
3
4
5
6
7
8
9
10
11
12
13
async def 问候多人(名字列表: List[str]) -> None:
    """
    向多个人依次问候
    
    示例：
        await 问候多人(["小明", "小红", "小李"])
        输出：
        你好，小明！
        你好，小红！
        你好，小李！
    """
    # 你的代码
    pass

练习2.3：模拟下载 ⭐⭐⭐

要求：编写一个异步函数，模拟下载文件（假设每MB需要0.5秒）。

1
2
3
4
5
6
7
8
9
10
11
12
13
async def 下载文件(文件名: str, 大小MB: int) -> Dict[str, any]:
    """
    模拟下载文件
    
    参数：
        文件名：文件名称
        大小MB：文件大小（MB）
    
    返回：
        包含文件名、大小、耗时的字典
    """
    # 你的代码
    pass

第3课练习：并发执行

练习3.1：批量查询天气 ⭐⭐

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
async def 查询天气(城市: str) -> Dict[str, any]:
    """
    查询单个城市的天气（模拟）
    
    返回：
        {"城市": "北京", "温度": 25, "天气": "晴天"}
    """
    # 你的代码
    pass

async def 批量查询天气(城市列表: List[str]) -> List[Dict]:
    """
    批量查询多个城市的天气
    """
    # 你的代码
    pass

练习3.2：限流下载器 ⭐⭐⭐

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

1
2
3
4
5
6
7
8
9
10
async def 限流下载器(文件列表: List[str], 最大并发: int = 3) -> None:
    """
    限流下载器，控制并发数量
    
    参数：
        文件列表：要下载的文件列表
        最大并发：最多同时下载的文件数
    """
    # 你的代码
    pass

练习3.3：超时重试 ⭐⭐⭐

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
async def 执行任务(
    任务名: str,
    超时秒数: int = 5,
    最大重试: int = 3
) -> Optional[str]:
    """
    执行任务，支持超时和重试
    
    参数：
        任务名：任务名称
        超时秒数：超时时间
        最大重试：最大重试次数
    
    返回：
        任务结果，失败返回None
    """
    # 你的代码
    pass

第4课练习：异步网络请求

练习4.1：批量图片下载器 ⭐⭐⭐

要求：编写一个异步图片下载器，同时下载多张图片。

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
async def 下载图片(url: str, 保存路径: str) -> bool:
    """
    下载单张图片
    
    参数：
        url：图片URL
        保存路径：保存位置
    
    返回：
        是否成功
    """
    # 你的代码
    pass

async def 批量下载图片(图片列表: List[tuple]) -> Dict[str, int]:
    """
    批量下载图片
    
    参数：
        图片列表：[(url1, path1), (url2, path2), ...]
    
    返回：
        {"成功": 10, "失败": 2}
    """
    # 你的代码
    pass

练习4.2：API监控 ⭐⭐⭐

要求：编写一个程序，定时检查多个API的可用性。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
async def 检查API(url: str) -> Dict[str, any]:
    """
    检查API是否可用
    
    返回：
        {"url": "...", "可用": True, "响应时间": 0.5}
    """
    # 你的代码
    pass

async def 监控API(url列表: List[str], 检查间隔: int = 60) -> None:
    """
    持续监控API
    
    参数：
        url列表：要监控的API列表
        检查间隔：检查间隔（秒）
    """
    # 你的代码
    pass

练习4.3：智能爬虫 ⭐⭐⭐⭐

要求：实现一个带重试、限流、错误处理的网页爬虫。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
async def 智能爬取(
    url: str,
    最大重试: int = 3,
    超时秒数: int = 10
) -> Optional[str]:
    """
    智能爬取网页
    
    功能：
        - 自动重试
        - 超时控制
        - 错误处理
    """
    # 你的代码
    pass

第5课练习：异步文件和数据库

练习5.1：日志分析器 ⭐⭐⭐

要求：编写一个程序，分析多个日志文件，统计各类日志级别的数量。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
async def 分析日志(文件路径: str) -> Dict[str, int]:
    """
    分析日志文件
    
    返回：
        {"INFO": 100, "WARNING": 20, "ERROR": 5}
    """
    # 你的代码
    pass

async def 批量分析日志(文件列表: List[str]) -> Dict[str, int]:
    """
    批量分析多个日志文件
    
    返回：
        汇总的统计结果
    """
    # 你的代码
    pass

练习5.2：CSV转JSON ⭐⭐⭐

要求：批量将CSV文件转换为JSON文件。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
async def CSV转JSON(csv文件: str, json文件: str) -> bool:
    """
    将CSV文件转换为JSON文件
    
    返回：
        是否成功
    """
    # 你的代码
    pass

async def 批量转换(文件对列表: List[tuple]) -> Dict[str, int]:
    """
    批量转换CSV到JSON
    
    参数：
        文件对列表：[(csv1, json1), (csv2, json2), ...]
    
    返回：
        {"成功": 10, "失败": 0}
    """
    # 你的代码
    pass

练习5.3：数据库备份 ⭐⭐⭐⭐

要求：实现一个异步数据库备份工具。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
async def 备份数据库(
    源数据库: str,
    目标数据库: str,
    表列表: Optional[List[str]] = None
) -> bool:
    """
    备份数据库
    
    参数：
        源数据库：源数据库路径
        目标数据库：目标数据库路径
        表列表：要备份的表（None表示全部）
    
    返回：
        是否成功
    """
    # 你的代码
    pass

第6课练习：陷阱和最佳实践

练习6.1：找出错误 ⭐⭐

问题：下面的代码有什么问题？如何修复？

1
2
3
4
5
6
7
8
9
10
11
12
async def 有问题的代码():
    import time
    
    # 问题1
    任务1 = 异步函数1()
    
    # 问题2
    time.sleep(2)
    
    # 问题3
    结果 = await 任务1
    return 结果

练习6.2：优化代码 ⭐⭐⭐

问题：优化下面的代码，提高性能：

1
2
3
4
5
6
7
8
9
10
11
async def 需要优化的代码():
    结果列表 = []
    
    for i in range(100):
        # 每次都创建新session
        async with aiohttp.ClientSession() as session:
            async with session.get(f"https://api.com/{i}") as response:
                结果 = await response.json()
                结果列表.append(结果)
    
    return 结果列表

练习6.3：添加错误处理 ⭐⭐⭐

问题：为下面的代码添加完善的错误处理：

1
2
3
4
async def 需要错误处理的代码(url: str):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.json()

综合练习：实战项目

项目1：新闻聚合器 ⭐⭐⭐⭐

要求：实现一个新闻聚合器，从多个新闻源抓取新闻。

功能：

  1. 同时爬取多个新闻网站
  2. 提取标题、链接、时间
  3. 按时间排序
  4. 错误处理和重试
  5. 限制并发数量

项目2：文件批处理系统 ⭐⭐⭐⭐

要求：实现一个文件批处理系统。

功能：

  1. 监控指定目录
  2. 自动处理新文件
  3. 支持多种文件格式
  4. 错误处理和日志
  5. 性能统计

项目3：API压力测试工具 ⭐⭐⭐⭐⭐

要求：实现一个API压力测试工具。

功能：

  1. 并发发送请求
  2. 统计响应时间
  3. 错误率统计
  4. 生成测试报告
  5. 支持多种请求方式

学习建议

  1. 循序渐进：从简单到复杂
  2. 动手实践：每题都要自己写代码
  3. 查阅文档：遇到不懂的查官方文档
  4. 对比答案：写完后对比参考答案
  5. 举一反三：尝试修改和扩展

提交作业

完成练习后，可以：

  1. 运行代码，确保正确性
  2. 添加注释，解释思路
  3. 优化代码，提高性能
  4. 分享给其他学习者

祝你学习愉快！ 🎉

练习答案.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
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
"""
Python异步编程练习题答案

运行方式：
    python 练习答案.py
"""

import asyncio
import time
from typing import List, Dict, Optional
import random


# ============================================
# 第2课练习答案
# ============================================

async def 倒计时(秒数: int) -> None:
    """
    练习2.1：异步倒计时
    从指定秒数倒数到0
    """
    for i in range(秒数, 0, -1):
        print(f"{i}...")
        await asyncio.sleep(1)
    print("时间到！")


async def 问候多人(名字列表: List[str]) -> None:
    """
    练习2.2：异步问候
    向多个人依次问候，每次间隔1秒
    """
    for 名字 in 名字列表:
        print(f"你好，{名字}！")
        await asyncio.sleep(1)


async def 下载文件(文件名: str, 大小MB: int) -> Dict[str, any]:
    """
    练习2.3：模拟下载
    模拟下载文件（每MB需要0.5秒）
    """
    print(f"📥 开始下载：{文件名} ({大小MB}MB)")

    开始时间 = time.time()
    下载时间 = 大小MB * 0.5
    await asyncio.sleep(下载时间)
    耗时 = time.time() - 开始时间

    print(f"✅ 下载完成：{文件名}")

    return {
        "文件名": 文件名,
        "大小": 大小MB,
        "耗时": 耗时
    }


# ============================================
# 第3课练习答案
# ============================================

async def 查询天气(城市: str) -> Dict[str, any]:
    """
    练习3.1：查询单个城市的天气（模拟）
    """
    # 模拟网络延迟
    await asyncio.sleep(random.uniform(0.3, 0.8))

    # 模拟天气数据
    return {
        "城市": 城市,
        "温度": random.randint(15, 30),
        "天气": random.choice(["晴天", "多云", "小雨", "阴天"])
    }


async def 批量查询天气(城市列表: List[str]) -> List[Dict]:
    """
    练习3.1：批量查询多个城市的天气
    """
    print(f"🌤️  查询 {len(城市列表)} 个城市的天气...")

    # 同时查询所有城市
    任务列表 = [查询天气(城市) for 城市 in 城市列表]
    结果列表 = await asyncio.gather(*任务列表)

    return 结果列表


async def 限流下载器(文件列表: List[str], 最大并发: int = 3) -> None:
    """
    练习3.2：限流下载器
    控制最多同时下载的文件数
    """
    信号量 = asyncio.Semaphore(最大并发)

    async def 受限下载(文件名: str):
        async with 信号量:
            print(f"📥 开始下载：{文件名}")
            await asyncio.sleep(1)  # 模拟下载
            print(f"✅ 完成下载：{文件名}")
            return 文件名

    print(f"🚀 开始下载（最多同时{最大并发}个）...")
    任务列表 = [受限下载(文件) for 文件 in 文件列表]
    await asyncio.gather(*任务列表)
    print("✅ 全部下载完成！")


async def 执行任务(
    任务名: str,
    超时秒数: int = 5,
    最大重试: int = 3
) -> Optional[str]:
    """
    练习3.3：执行任务，支持超时和重试
    """

    async def 实际任务():
        """模拟可能失败的任务"""
        await asyncio.sleep(random.uniform(0.5, 2))
        if random.random() < 0.3:  # 30%概率失败
            raise Exception("任务执行失败")
        return f"{任务名}完成"

    for 尝试次数 in range(最大重试):
        try:
            print(f"  尝试 {尝试次数 + 1}/{最大重试}: {任务名}")

            # 执行任务，设置超时
            结果 = await asyncio.wait_for(实际任务(), timeout=超时秒数)

            print(f"  ✅ {任务名}成功")
            return 结果

        except asyncio.TimeoutError:
            print(f"  ❌ {任务名}超时")

        except Exception as e:
            print(f"  ❌ {任务名}失败：{e}")

        # 如果不是最后一次尝试，等待后重试
        if 尝试次数 < 最大重试 - 1:
            等待时间 = (尝试次数 + 1) * 0.5
            print(f"  ⏳ 等待 {等待时间}秒 后重试...")
            await asyncio.sleep(等待时间)

    print(f"  ❌ {任务名}最终失败")
    return None


# ============================================
# 第4课练习答案
# ============================================

async def 下载图片(url: str, 保存路径: str) -> bool:
    """
    练习4.1：下载单张图片（模拟）
    """
    try:
        print(f"📥 下载图片：{url}")

        # 模拟下载
        await asyncio.sleep(random.uniform(0.5, 1.5))

        # 模拟保存文件
        # 实际项目中这里会用 aiofiles 和 aiohttp
        print(f"💾 保存到：{保存路径}")

        return True

    except Exception as e:
        print(f"❌ 下载失败：{url} - {e}")
        return False


async def 批量下载图片(图片列表: List[tuple]) -> Dict[str, int]:
    """
    练习4.1：批量下载图片
    """
    print(f"🚀 开始下载 {len(图片列表)} 张图片...")

    任务列表 = [下载图片(url, path) for url, path in 图片列表]
    结果列表 = await asyncio.gather(*任务列表)

    成功数 = sum(1 for r in 结果列表 if r)
    失败数 = len(结果列表) - 成功数

    return {"成功": 成功数, "失败": 失败数}


async def 检查API(url: str) -> Dict[str, any]:
    """
    练习4.2：检查API是否可用
    """
    try:
        开始时间 = time.time()

        # 模拟API请求
        await asyncio.sleep(random.uniform(0.1, 0.5))

        响应时间 = time.time() - 开始时间

        # 模拟90%成功率
        可用 = random.random() < 0.9

        return {
            "url": url,
            "可用": 可用,
            "响应时间": 响应时间
        }

    except Exception as e:
        return {
            "url": url,
            "可用": False,
            "错误": str(e)
        }


async def 监控API(url列表: List[str], 检查间隔: int = 60, 检查次数: int = 3) -> None:
    """
    练习4.2：持续监控API（简化版，只检查3次）
    """
    print(f"🔍 开始监控 {len(url列表)} 个API...")

    for 轮次 in range(检查次数):
        print(f"\n第 {轮次 + 1} 轮检查：")

        任务列表 = [检查API(url) for url in url列表]
        结果列表 = await asyncio.gather(*任务列表)

        for 结果 in 结果列表:
            状态 = "✅" if 结果["可用"] else "❌"
            if 结果["可用"]:
                print(f"  {状态} {结果['url']} - 响应时间: {结果['响应时间']:.2f}秒")
            else:
                print(f"  {状态} {结果['url']} - 不可用")

        if 轮次 < 检查次数 - 1:
            print(f"\n⏳ 等待 {检查间隔}秒...")
            await asyncio.sleep(检查间隔)


async def 智能爬取(
    url: str,
    最大重试: int = 3,
    超时秒数: int = 10
) -> Optional[str]:
    """
    练习4.3：智能爬取网页
    带重试、超时、错误处理
    """

    async def 爬取():
        """实际爬取操作"""
        # 模拟网络请求
        await asyncio.sleep(random.uniform(0.5, 2))

        # 模拟80%成功率
        if random.random() < 0.8:
            return f"网页内容：{url}"
        else:
            raise Exception("网络错误")

    for 尝试次数 in range(最大重试):
        try:
            print(f"  🕷️  尝试 {尝试次数 + 1}/{最大重试}: {url}")

            # 执行爬取，设置超时
            结果 = await asyncio.wait_for(爬取(), timeout=超时秒数)

            print(f"  ✅ 爬取成功")
            return 结果

        except asyncio.TimeoutError:
            print(f"  ❌ 超时")

        except Exception as e:
            print(f"  ❌ 失败：{e}")

        # 重试前等待
        if 尝试次数 < 最大重试 - 1:
            等待时间 = (尝试次数 + 1) * 1.0
            await asyncio.sleep(等待时间)

    print(f"  ❌ 爬取最终失败")
    return None


# ============================================
# 第5课练习答案（简化版，使用模拟数据）
# ============================================

async def 分析日志(文件路径: str) -> Dict[str, int]:
    """
    练习5.1：分析日志文件
    """
    # 模拟读取日志文件
    await asyncio.sleep(0.2)

    # 模拟统计结果
    return {
        "INFO": random.randint(50, 100),
        "WARNING": random.randint(10, 30),
        "ERROR": random.randint(0, 10)
    }


async def 批量分析日志(文件列表: List[str]) -> Dict[str, int]:
    """
    练习5.1：批量分析多个日志文件
    """
    print(f"📝 分析 {len(文件列表)} 个日志文件...")

    任务列表 = [分析日志(文件) for 文件 in 文件列表]
    结果列表 = await asyncio.gather(*任务列表)

    # 汇总统计
    总统计 = {"INFO": 0, "WARNING": 0, "ERROR": 0}
    for 结果 in 结果列表:
        for 级别, 数量 in 结果.items():
            总统计[级别] += 数量

    return 总统计


# ============================================
# 第6课练习答案
# ============================================

# 练习6.1：找出错误
"""
问题1：任务1 = 异步函数1()
    - 忘记使用 await，只得到协程对象

问题2：time.sleep(2)
    - 使用了阻塞操作，应该用 asyncio.sleep(2)

问题3：虽然使用了await，但任务1从未真正执行

正确代码：
"""


async def 修复后的代码():
    # 修复1：使用 await
    结果1 = await 异步函数1()

    # 修复2：使用异步sleep
    await asyncio.sleep(2)

    return 结果1


# 练习6.2：优化代码
"""
问题：每次循环都创建新的session，浪费资源

优化后的代码：
"""


async def 优化后的代码():
    结果列表 = []

    # 复用session
    # async with aiohttp.ClientSession() as session:
    #     # 批量创建任务
    #     任务列表 = [session.get(f"https://api.com/{i}") for i in range(100)]
    #     响应列表 = await asyncio.gather(*任务列表)
    #
    #     # 处理响应
    #     for 响应 in 响应列表:
    #         async with 响应:
    #             结果 = await 响应.json()
    #             结果列表.append(结果)

    return 结果列表


# 练习6.3：添加错误处理
"""
优化后的代码：
"""


async def 添加错误处理后的代码(url: str) -> Optional[Dict]:
    try:
        # async with aiohttp.ClientSession() as session:
        #     async with session.get(url, timeout=10) as response:
        #         if response.status == 200:
        #             return await response.json()
        #         else:
        #             print(f"❌ 状态码错误：{response.status}")
        #             return None
        pass

    except asyncio.TimeoutError:
        print(f"❌ 请求超时：{url}")
        return None

    # except aiohttp.ClientError as e:
    #     print(f"❌ 网络错误：{e}")
    #     return None

    except Exception as e:
        print(f"❌ 未知错误：{e}")
        return None


# ============================================
# 测试所有练习
# ============================================

async def 测试第2课():
    """测试第2课练习"""
    print("\n" + "=" * 50)
    print("📚 第2课练习测试")
    print("=" * 50)

    # 测试2.1
    print("\n练习2.1：异步倒计时")
    await 倒计时(3)

    # 测试2.2
    print("\n练习2.2：异步问候")
    await 问候多人(["小明", "小红"])

    # 测试2.3
    print("\n练习2.3：模拟下载")
    结果 = await 下载文件("test.pdf", 4)
    print(f"下载结果：{结果}")


async def 测试第3课():
    """测试第3课练习"""
    print("\n" + "=" * 50)
    print("📚 第3课练习测试")
    print("=" * 50)

    # 测试3.1
    print("\n练习3.1：批量查询天气")
    城市列表 = ["北京", "上海", "广州"]
    结果 = await 批量查询天气(城市列表)
    for 天气 in 结果:
        print(f"  {天气['城市']}: {天气['温度']}°C, {天气['天气']}")

    # 测试3.2
    print("\n练习3.2：限流下载器")
    文件列表 = [f"文件{i}.txt" for i in range(1, 6)]
    await 限流下载器(文件列表, 最大并发=2)

    # 测试3.3
    print("\n练习3.3：超时重试")
    await 执行任务("测试任务", 超时秒数=3, 最大重试=2)


async def 测试第4课():
    """测试第4课练习"""
    print("\n" + "=" * 50)
    print("📚 第4课练习测试")
    print("=" * 50)

    # 测试4.1
    print("\n练习4.1：批量下载图片")
    图片列表 = [
        ("https://example.com/1.jpg", "img1.jpg"),
        ("https://example.com/2.jpg", "img2.jpg"),
        ("https://example.com/3.jpg", "img3.jpg"),
    ]
    统计 = await 批量下载图片(图片列表)
    print(f"下载统计：{统计}")

    # 测试4.2
    print("\n练习4.2：API监控")
    url列表 = [
        "https://api1.com/health",
        "https://api2.com/health",
    ]
    await 监控API(url列表, 检查间隔=2, 检查次数=2)

    # 测试4.3
    print("\n练习4.3：智能爬取")
    await 智能爬取("https://example.com", 最大重试=2)


async def 测试第5课():
    """测试第5课练习"""
    print("\n" + "=" * 50)
    print("📚 第5课练习测试")
    print("=" * 50)

    # 测试5.1
    print("\n练习5.1：批量分析日志")
    文件列表 = ["log1.txt", "log2.txt", "log3.txt"]
    统计 = await 批量分析日志(文件列表)
    print(f"日志统计：{统计}")


async def main():
    """主程序"""
    print("🎓 Python异步编程练习答案")
    print("=" * 50)

    # 运行所有测试
    await 测试第2课()
    await 测试第3课()
    await 测试第4课()
    await 测试第5课()

    print("\n" + "=" * 50)
    print("🎉 所有练习测试完成！")
    print("=" * 50)
    print("""
💡 学习建议：
    1. 对比自己的答案和参考答案
    2. 理解每个解决方案的思路
    3. 尝试优化和改进代码
    4. 应用到实际项目中
    
📚 继续学习：
    1. 阅读官方文档
    2. 研究优秀开源项目
    3. 实践更多项目
    4. 分享学习心得
    
🚀 加油！
    """)


if __name__ == "__main__":
    asyncio.run(main())