08_Python异步编程练习题

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

本课程包含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 countdown(seconds: int) -> None:
    """
    从指定秒数倒数到0
    
    示例：
        await countdown(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 greet_multiple_people(name_list: List[str]) -> None:
    """
    向多个人依次问候
    
    示例：
        await greet_multiple_people(["小明", "小红", "小李"])
        输出：
        你好，小明！
        你好，小红！
        你好，小李！
    """
    # 你的代码
    pass

练习2.3：模拟下载 ⭐⭐⭐

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

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

第3课练习：并发执行

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

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

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

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

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

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

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

练习3.3：超时重试 ⭐⭐⭐

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
async def execute_task(
    task_name: str,
    timeout_seconds: int = 5,
    max_retry: int = 3
) -> Optional[str]:
    """
    执行任务，支持超时和重试
    
    param：
        任务名：任务名称
        timeout_seconds：超时时间
        max_retry：最大重试次数
    
    返回：
        任务结果，失败返回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 download_image(url: str, save_path: str) -> bool:
    """
    下载单张图片
    
    param：
        url：图片URL
        保存路径：保存位置
    
    返回：
        是否成功
    """
    # 你的代码
    pass

async def batch_download_images(image_list: List[tuple]) -> Dict[str, int]:
    """
    batch_download_images
    
    param：
        image_list：[(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 check_api(url: str) -> Dict[str, any]:
    """
    检查API是否可用
    
    返回：
        {"url": "...", "available": True, "response_time": 0.5}
    """
    # 你的代码
    pass

async def monitor_api(url_list: List[str], check_interval: int = 60) -> None:
    """
    持续监控API
    
    param：
        url_list：要监控的API列表
        check_interval：check_interval（秒）
    """
    # 你的代码
    pass

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

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
async def smart_crawl(
    url: str,
    max_retry: int = 3,
    timeout_seconds: 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 analyze_log(file_path: str) -> Dict[str, int]:
    """
    分析日志文件
    
    返回：
        {"INFO": 100, "WARNING": 20, "ERROR": 5}
    """
    # 你的代码
    pass

async def batch_analyze_logs(file_list: 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_to_json(csv_file: str, json_file: str) -> bool:
    """
    将CSV文件转换为JSON文件
    
    返回：
        是否成功
    """
    # 你的代码
    pass

async def batch_convert(file_pair_list: List[tuple]) -> Dict[str, int]:
    """
    批量转换CSV到JSON
    
    param：
        文件对列表：[(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 backup_database(
    source_database: str,
    target_database: str,
    table_list: Optional[List[str]] = None
) -> bool:
    """
    备份数据库
    
    param：
        源数据库：源数据库路径
        目标数据库：目标数据库路径
        table_list：要备份的表（None表示全部）
    
    返回：
        是否成功
    """
    # 你的代码
    pass

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

练习6.1：找出错误 ⭐⭐

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

1
2
3
4
5
6
7
8
9
10
11
12
async def problematic_code():
    import time
    
    # 问题1
    task1 = async_function1()
    
    # 问题2
    time.sleep(2)
    
    # 问题3
    result = await task1
    return result

练习6.2：优化代码 ⭐⭐⭐

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

1
2
3
4
5
6
7
8
9
10
11
async def code_needing_optimization():
    result_list = []
    
    for i in range(100):
        # 每次都创建新session
        async with aiohttp.ClientSession() as session:
            async with session.get(f"https://api.com/{i}") as response:
                result = await response.json()
                result_list.append(result)
    
    return result_list

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

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

1
2
3
4
async def code_needing_error_handling(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 countdown(seconds: int) -> None:
    """
    练习2.1：异步倒计时
    从指定秒数倒数到0
    """
    for i in range(seconds, 0, -1):
        print(f"{i}...")
        await asyncio.sleep(1)
    print("时间到！")


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


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

    start_time = time.time()
    download_time = size_mb * 0.5
    await asyncio.sleep(download_time)
    duration = time.time() - start_time

    print(f"✅ 下载完成：{filename}")

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


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

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

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


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

    # 同时查询所有城市
    task_list = [query_weather(city) for city in city_list]
    result_list = await asyncio.gather(*task_list)

    return result_list


async def throttled_downloader(file_list: List[str], max_concurrency: int = 3) -> None:
    """
    练习3.2：throttled_downloader
    控制最多同时下载的文件数
    """
    semaphore = asyncio.Semaphore(max_concurrency)

    async def limited_download(filename: str):
        async with semaphore:
            print(f"📥 开始下载：{filename}")
            await asyncio.sleep(1)  # 模拟下载
            print(f"✅ 完成下载：{filename}")
            return filename

    print(f"🚀 开始下载（最多同时{max_concurrency}个）...")
    task_list = [throttled_download(file) for file in file_list]
    await asyncio.gather(*task_list)
    print("✅ 全部下载完成！")


async def execute_task(
    task_name: str,
    timeout_seconds: int = 5,
    max_retry: int = 3
) -> Optional[str]:
    """
    练习3.3：执行任务，支持超时和重试
    """

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

    for attempt_count in range(max_retry):
        try:
            print(f"  尝试 {attempt_count + 1}/{max_retry}: {任务名}")

            # 执行任务，设置超时
            result = await asyncio.wait_for(actual_task(), timeout=timeout_seconds)

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

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

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

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

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


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

async def download_image(url: str, save_path: str) -> bool:
    """
    练习4.1：下载单张图片（模拟）
    """
    try:
        print(f"📥 download_image：{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 batch_download_images(image_list: List[tuple]) -> Dict[str, int]:
    """
    练习4.1：batch_download_images
    """
    print(f"🚀 开始下载 {len(image_list)} 张图片...")

    task_list = [download_image(url, path) for url, path in image_list]
    result_list = await asyncio.gather(*task_list)

    success_count = sum(1 for r in result_list if r)
    failure_count = len(result_list) - success_count

    return {"成功": success_count, "失败": failure_count}


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

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

        response_time = time.time() - start_time

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

        return {
            "url": url,
            "available": available,
            "response_time": response_time
        }

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


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

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

        task_list = [check_api(url) for url in url_list]
        result_list = await asyncio.gather(*task_list)

        for result in result_list:
            status = "✅" if result["available"] else "❌"
            if result["available"]:
                print(f"  {status} {result['url']} - response_time: {result['response_time']:.2f}秒")
            else:
                print(f"  {status} {result['url']} - 不可用")

        if round < check_count - 1:
            print(f"\n⏳ 等待 {check_interval}秒...")
            await asyncio.sleep(check_interval)


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

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

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

    for attempt_count in range(max_retry):
        try:
            print(f"  🕷️  尝试 {attempt_count + 1}/{max_retry}: {url}")

            # 执行爬取，设置超时
            result = await asyncio.wait_for(crawl(), timeout=timeout_seconds)

            print(f"  ✅ 爬取成功")
            return result

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

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

        # 重试前等待
        if attempt_count < max_retry - 1:
            wait_time = (attempt_count + 1) * 1.0
            await asyncio.sleep(wait_time)

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


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

async def analyze_log(file_path: 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 batch_analyze_logs(file_list: List[str]) -> Dict[str, int]:
    """
    练习5.1：批量分析多个日志文件
    """
    print(f"📝 分析 {len(file_list)} 个日志文件...")

    task_list = [analyze_log(file) for file in file_list]
    result_list = await asyncio.gather(*task_list)

    # 汇总统计
    total_stats = {"INFO": 0, "WARNING": 0, "ERROR": 0}
    for result in result_list:
        for level, count in result.items():
            total_stats[level] += count

    return total_stats


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

# 练习6.1：找出错误
"""
问题1：task1 = async_function1()
    - 忘记使用 await，只得到协程对象

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

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

正确代码：
"""


async def fixed_code():
    # 修复1：使用 await
    result1 = await async_function1()

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

    return result1


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

优化后的代码：
"""


async def optimized_code():
    result_list = []

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

    return result_list


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


async def code_with_error_handling(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 test_lesson2():
    """测试第2课练习"""
    print("\n" + "=" * 50)
    print("📚 第2课练习测试")
    print("=" * 50)

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

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

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


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

    # 测试3.1
    print("\n练习3.1：batch_query_weather")
    city_list = ["北京", "上海", "广州"]
    result = await batch_query_weather(city_list)
    for weather in result:
        print(f"  {weather['city']}: {weather['温度']}°C, {weather['weather']}")

    # 测试3.2
    print("\n练习3.2：throttled_downloader")
    file_list = [f"file{i}.txt" for i in range(1, 6)]
    await throttled_downloader(file_list, max_concurrency=2)

    # 测试3.3
    print("\n练习3.3：超时重试")
    await execute_task("测试任务", timeout_seconds=3, max_retry=2)


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

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

    # 测试4.2
    print("\n练习4.2：API监控")
    url_list = [
        "https://api1.com/health",
        "https://api2.com/health",
    ]
    await monitor_api(url_list, check_interval=2, check_count=2)

    # 测试4.3
    print("\n练习4.3：smart_crawl")
    await smart_crawl("https://example.com", max_retry=2)


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

    # 测试5.1
    print("\n练习5.1：batch_analyze_logs")
    file_list = ["log1.txt", "log2.txt", "log3.txt"]
    stats = await batch_analyze_logs(file_list)
    print(f"日志统计：{统计}")


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

    # 运行所有测试
    await test_lesson2()
    await test_lesson3()
    await test_lesson4()
    await test_lesson5()

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


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