10个实用Python自动化脚本：从文件整理到数据备份
每天重复做同样的事情——整理下载文件夹、备份重要文件、批量重命名图片、清理桌面……这些琐碎的任务单个看起来不值一提，但加在一起，每周可能浪费你好几个小时。
我之前也是这样。后来开始用Python把这些重复操作自动化，效果立竿见影。今天分享10个我实际在用的自动化脚本，每个都是解决真实痛点的，不是那种”看起来很酷但从来不会用”的demo。
1. 智能文件整理器
下载文件夹永远是重灾区。这个脚本按文件类型自动归类：
import shutil
import os
from pathlib import Path
FILE_CATEGORIES = {
"文档": [".pdf", ".docx", ".doc", ".txt", ".xlsx", ".csv", ".pptx"],
"图片": [".jpg", ".jpeg", ".png", ".gif", ".webp", ".svg", ".bmp"],
"视频": [".mp4", ".mov", ".avi", ".mkv", ".webm"],
"音频": [".mp3", ".wav", ".flac", ".aac", ".ogg"],
"压缩包": [".zip", ".rar", ".7z", ".tar", ".gz"],
"代码": [".py", ".js", ".ts", ".html", ".css", ".json", ".yaml"],
"安装包": [".dmg", ".exe", ".msi", ".deb", ".rpm", ".pkg"],
}
def organize_downloads(folder_path):
"""按文件类型整理文件夹"""
folder = Path(folder_path)
organized_count = 0
for file in folder.iterdir():
if not file.is_file():
continue
# 找到文件所属分类
target_category = None
for category, extensions in FILE_CATEGORIES.items():
if file.suffix.lower() in extensions:
target_category = category
break
if not target_category:
target_category = "其他"
# 创建分类目录并移动文件
target_dir = folder / target_category
target_dir.mkdir(exist_ok=True)
target_path = target_dir / file.name
# 避免文件名冲突
if target_path.exists():
stem = file.stem
counter = 1
while target_path.exists():
target_path = target_dir / f"{stem}_{counter}{file.suffix}"
counter += 1
shutil.move(str(file), str(target_path))
organized_count += 1
print(f"整理完成！共移动 {organized_count} 个文件")
# 使用
organize_downloads(os.path.expanduser("~/Downloads"))
2. 自动备份脚本
重要数据定期备份，这个脚本支持增量备份（只备份变化的文件）：
import shutil
import hashlib
import json
from pathlib import Path
from datetime import datetime
BACKUP_MANIFEST = ".backup_manifest.json"
def get_file_hash(filepath):
"""计算文件MD5哈希"""
hasher = hashlib.md5()
with open(filepath, "rb") as f:
for chunk in iter(lambda: f.read(8192), b""):
hasher.update(chunk)
return hasher.hexdigest()
def backup_folder(source, backup_root):
"""增量备份文件夹"""
source_path = Path(source)
backup_root_path = Path(backup_root)
manifest_path = backup_root_path / BACKUP_MANIFEST
# 读取上次的备份记录
manifest = {}
if manifest_path.exists():
manifest = json.loads(manifest_path.read_text())
# 创建带时间戳的备份目录
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_path = backup_root_path / timestamp
backup_path.mkdir(parents=True, exist_ok=True)
new_manifest = {}
backed_up = 0
skipped = 0
for file in source_path.rglob("*"):
if not file.is_file():
continue
rel_path = str(file.relative_to(source_path))
file_hash = get_file_hash(file)
# 检查文件是否变化
if rel_path in manifest and manifest[rel_path] == file_hash:
skipped += 1
new_manifest[rel_path] = file_hash
continue
# 复制变化的文件
target = backup_path / rel_path
target.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(file, target)
new_manifest[rel_path] = file_hash
backed_up += 1
# 更新备份记录
manifest_path.write_text(json.dumps(new_manifest, indent=2))
print(f"备份完成！新增/更新 {backed_up} 个文件，跳过 {skipped} 个未变化文件")
print(f"备份位置: {backup_path}")
# 使用
backup_folder(
source=os.path.expanduser("~/Documents/重要项目"),
backup_root=os.path.expanduser("~/Backup")
)
3. 批量图片重命名
从相机或手机导出的图片名字都是IMG_0001这种，这个脚本按拍摄日期重命名：
from pathlib import Path
from datetime import datetime
from PIL import Image
import os
def rename_photos_by_date(folder_path):
"""按拍摄日期重命名照片"""
folder = Path(folder_path)
renamed = 0
for file in sorted(folder.iterdir()):
if not file.is_file():
continue
if file.suffix.lower() not in [".jpg", ".jpeg", ".png", ".heic"]:
continue
try:
img = Image.open(file)
exif_data = img._getexif()
if exif_data and 36867 in exif_data:
# 36867 是EXIF中的拍摄日期字段
date_str = exif_data[36867]
date = datetime.strptime(date_str, "%Y:%m:%d %H:%M:%S")
else:
# 没有EXIF信息，使用文件修改时间
date = datetime.fromtimestamp(file.stat().st_mtime)
new_name = date.strftime("%Y%m%d_%H%M%S") + file.suffix
new_path = file.parent / new_name
# 避免重名
counter = 1
while new_path.exists():
new_name = date.strftime("%Y%m%d_%H%M%S") + f"_{counter}" + file.suffix
new_path = file.parent / new_name
counter += 1
file.rename(new_path)
renamed += 1
except Exception as e:
print(f"处理失败: {file.name} - {e}")
print(f"重命名完成！共处理 {renamed} 张照片")
# 使用
rename_photos_by_date(os.path.expanduser("~/Pictures/待整理"))
4. 桌面清理助手
超过30天未使用的桌面文件自动归档：
import shutil
import os
import time
from pathlib import Path
def clean_desktop(desktop_path, archive_path, days_threshold=30):
"""清理桌面：将超过指定天数的文件移到归档目录"""
desktop = Path(desktop_path)
archive = Path(archive_path)
archive.mkdir(parents=True, exist_ok=True)
now = time.time()
threshold_seconds = days_threshold * 86400
moved = 0
for file in desktop.iterdir():
if not file.is_file():
continue
last_access = file.stat().st_atime
if (now - last_access) > threshold_seconds:
# 按原始日期归档
file_date = time.strftime(
"%Y-%m", time.localtime(file.stat().st_mtime)
)
target_dir = archive / file_date
target_dir.mkdir(exist_ok=True)
target = target_dir / file.name
if target.exists():
target = target_dir / f"{file.stem}_{int(now)}{file.suffix}"
shutil.move(str(file), str(target))
moved += 1
print(f"归档: {file.name} -> {target_dir.name}/")
print(f"\n清理完成！归档 {moved} 个文件")
# 使用
clean_desktop(
desktop_path=os.path.expanduser("~/Desktop"),
archive_path=os.path.expanduser("~/DesktopArchive"),
days_threshold=30
)
5. 网页截图批量工具
批量截取网页的长截图，适合做竞品分析或存档：
import asyncio
from playwright.async_api import async_playwright
from pathlib import Path
async def batch_screenshot(urls_file, output_dir):
"""批量截取网页截图"""
output = Path(output_dir)
output.mkdir(parents=True, exist_ok=True)
urls = Path(urls_file).read_text().strip().split("\n")
async with async_playwright() as p:
browser = await p.chromium.launch()
page = await browser.new_page(viewport={"width": 1440, "height": 900})
for url in urls:
url = url.strip()
if not url:
continue
try:
await page.goto(url, wait_until="networkidle")
# 等待页面完全加载
await page.wait_for_timeout(2000)
filename = url.replace("https://", "").replace("http://", "").replace("/", "_")[:50]
filepath = output / f"{filename}.png"
await page.screenshot(path=str(filepath), full_page=True)
print(f"截图完成: {filename}")
except Exception as e:
print(f"截图失败: {url} - {e}")
await browser.close()
print(f"\n全部完成！截图保存在: {output_dir}")
# 使用
asyncio.run(batch_screenshot("urls.txt", "./screenshots"))
6. CSV数据清洗工具
自动清洗CSV文件中的常见问题：空值、重复行、格式不统一：
import pandas as pd
from pathlib import Path
def clean_csv(input_path, output_path=None):
"""清洗CSV文件"""
input_file = Path(input_path)
if output_path is None:
output_path = input_file.parent / f"cleaned_{input_file.name}"
df = pd.read_csv(input_file)
original_rows = len(df)
print(f"原始数据: {original_rows} 行, {len(df.columns)} 列")
# 删除完全重复的行
df = df.drop_duplicates()
print(f"去重后: {len(df)} 行 (删除 {original_rows - len(df)} 行)")
# 清理列名（去除空格和特殊字符）
df.columns = df.columns.str.strip().str.lower().str.replace(" ", "_")
# 填充或删除空值
for col in df.columns:
null_count = df[col].isnull().sum()
if null_count > 0:
if df[col].dtype in ["int64", "float64"]:
df[col] = df[col].fillna(df[col].median())
else:
df[col] = df[col].fillna("未知")
print(f" 列 '{col}': 填充 {null_count} 个空值")
# 去除字符串列的前后空格
for col in df.select_dtypes(include=["object"]).columns:
df[col] = df[col].str.strip()
# 保存清洗后的数据
df.to_csv(output_path, index=False, encoding="utf-8-sig")
print(f"\n清洗完成！保存到: {output_path}")
# 使用
clean_csv("raw_data.csv")
7. 定时提醒脚本
在长时间工作时定时提醒休息：
import time
import subprocess
import platform
REMINDERS = [
"站起来活动一下！",
"喝杯水吧，你已经很久没喝水了",
"看看远处，放松一下眼睛",
"深呼吸三次，缓解肩颈紧张",
]
def send_notification(title, message):
"""发送系统通知"""
system = platform.system()
try:
if system == "Darwin": # macOS
subprocess.run([
"osascript", "-e",
f'display notification "{message}" with title "{title}"'
])
elif system == "Linux":
subprocess.run([
"notify-send", title, message
])
elif system == "Windows":
subprocess.run([
"powershell", "-Command",
f'[System.Reflection.Assembly]::LoadWithPartialName("System.Windows.Forms"); '
f'[System.Windows.Forms.MessageBox]::Show("{message}", "{title}")'
])
except Exception as e:
print(f"[通知] {title}: {message}")
def pomodoro_timer(work_minutes=25, break_minutes=5):
"""番茄钟定时器"""
cycle = 1
while True:
print(f"\n--- 第 {cycle} 个工作周期 ({work_minutes}分钟) ---")
time.sleep(work_minutes * 60)
reminder = REMINDERS[(cycle - 1) % len(REMINDERS)]
send_notification("休息时间到！", reminder)
print(f"--- 休息时间 ({break_minutes}分钟) ---")
time.sleep(break_minutes * 60)
send_notification("休息结束", "开始下一个工作周期吧！")
cycle += 1
# 使用
pomodoro_timer(work_minutes=25, break_minutes=5)
8. Markdown文件批量转换
把Markdown文件批量转换为HTML或PDF：
import markdown
from pathlib import Path
from jinja2 import Template
HTML_TEMPLATE = Template("""
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title></title>
<style>
body { font-family: -apple-system, sans-serif; max-width: 800px; margin: 0 auto; padding: 2rem; line-height: 1.8; }
code { background: #f4f4f4; padding: 0.2rem 0.4rem; border-radius: 3px; }
pre { background: #f4f4f4; padding: 1rem; border-radius: 5px; overflow-x: auto; }
blockquote { border-left: 4px solid #ddd; margin: 0; padding-left: 1rem; color: #666; }
</style>
</head>
<body>
<article>
<h1>Python Asyncio的10个常见陷阱及解决方案</h1>
<div class="post-byline">
<span class="post-date">2026-06-02</span>
<span class="post-author">编译员：大D</span>
<span class="post-tags">
<a href="/tags/代码产品/" class="tag">代码产品</a>
</span>
</div>
<p>Python的asyncio库让异步编程变得前所未有的简单，但简单并不意味着容易。在实际项目中，很多开发者会掉进一些看似不起眼的陷阱，导致程序性能下降、死锁甚至崩溃。今天我把这几年踩过的坑和解决方案整理出来，希望能帮你少走弯路。</p>
<h2 id="陷阱1在异步函数中调用阻塞io">陷阱1：在异步函数中调用阻塞IO</h2>
<p>这是最常见的错误。很多开发者在async函数里直接使用<code class="language-plaintext highlighter-rouge">requests.get()</code>、<code class="language-plaintext highlighter-rouge">open()</code>等同步IO操作，结果整个事件循环被阻塞，其他协程全部卡住。</p>
<div class="language-python highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="c1"># 错误示范
</span><span class="k">async</span> <span class="k">def</span> <span class="nf">fetch_data</span><span class="p">():</span>
<span class="n">response</span> <span class="o">=</span> <span class="n">requests</span><span class="p">.</span><span class="n">get</span><span class="p">(</span><span class="s">"https://api.example.com/data"</span><span class="p">)</span> <span class="c1"># 阻塞整个事件循环
</span> <span class="k">return</span> <span class="n">response</span><span class="p">.</span><span class="n">json</span><span class="p">()</span>
<span class="c1"># 正确做法：使用aiohttp
</span><span class="kn">import</span> <span class="nn">aiohttp</span>
<span class="k">async</span> <span class="k">def</span> <span class="nf">fetch_data</span><span class="p">():</span>
<span class="k">async</span> <span class="k">with</span> <span class="n">aiohttp</span><span class="p">.</span><span class="n">ClientSession</span><span class="p">()</span> <span class="k">as</span> <span class="n">session</span><span class="p">:</span>
<span class="k">async</span> <span class="k">with</span> <span class="n">session</span><span class="p">.</span><span class="n">get</span><span class="p">(</span><span class="s">"https://api.example.com/data"</span><span class="p">)</span> <span class="k">as</span> <span class="n">response</span><span class="p">:</span>
<span class="k">return</span> <span class="k">await</span> <span class="n">response</span><span class="p">.</span><span class="n">json</span><span class="p">()</span>
<span class="c1"># 或者用run_in_executor包装同步调用
</span><span class="kn">import</span> <span class="nn">asyncio</span>
<span class="kn">from</span> <span class="nn">functools</span> <span class="kn">import</span> <span class="n">partial</span>
<span class="k">async</span> <span class="k">def</span> <span class="nf">fetch_data</span><span class="p">():</span>
<span class="n">loop</span> <span class="o">=</span> <span class="n">asyncio</span><span class="p">.</span><span class="n">get_running_loop</span><span class="p">()</span>
<span class="n">result</span> <span class="o">=</span> <span class="k">await</span> <span class="n">loop</span><span class="p">.</span><span class="n">run_in_executor</span><span class="p">(</span>
<span class="bp">None</span><span class="p">,</span>
<span class="n">partial</span><span class="p">(</span><span class="n">requests</span><span class="p">.</span><span class="n">get</span><span class="p">,</span> <span class="s">"https://api.example.com/data"</span><span class="p">)</span>
<span class="p">)</span>
<span class="k">return</span> <span class="n">result</span><span class="p">.</span><span class="n">json</span><span class="p">()</span>
</code></pre></div></div>
<p><strong>关键原则</strong>：在async函数中，任何可能超过几毫秒的IO操作都应该用异步版本或放到线程池中执行。</p>
<h2 id="陷阱2忘记await">陷阱2：忘记await</h2>
<p>忘记加<code class="language-plaintext highlighter-rouge">await</code>是asyncio中最隐蔽的bug。代码不会报错，但协程不会被执行，你只会得到一个coroutine对象。</p>
<div class="language-python highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="c1"># 错误示范
</span><span class="k">async</span> <span class="k">def</span> <span class="nf">main</span><span class="p">():</span>
<span class="n">task</span> <span class="o">=</span> <span class="n">asyncio</span><span class="p">.</span><span class="n">create_task</span><span class="p">(</span><span class="n">some_work</span><span class="p">())</span> <span class="c1"># 创建了任务但没有等待
</span> <span class="c1"># some_work可能还没执行完，main就结束了
</span>
<span class="c1"># 正确做法
</span><span class="k">async</span> <span class="k">def</span> <span class="nf">main</span><span class="p">():</span>
<span class="n">task</span> <span class="o">=</span> <span class="n">asyncio</span><span class="p">.</span><span class="n">create_task</span><span class="p">(</span><span class="n">some_work</span><span class="p">())</span>
<span class="n">result</span> <span class="o">=</span> <span class="k">await</span> <span class="n">task</span> <span class="c1"># 等待任务完成
</span></code></pre></div></div>
<p>更隐蔽的情况是在函数调用链中遗漏await：</p>
<div class="language-python highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="c1"># 错误：中间某层忘记await
</span><span class="k">async</span> <span class="k">def</span> <span class="nf">process</span><span class="p">(</span><span class="n">data</span><span class="p">):</span>
<span class="n">cleaned</span> <span class="o">=</span> <span class="n">clean_data</span><span class="p">(</span><span class="n">data</span><span class="p">)</span> <span class="c1"># 忘记await！clean_data是协程
</span> <span class="k">return</span> <span class="n">cleaned</span>
<span class="c1"># 正确
</span><span class="k">async</span> <span class="k">def</span> <span class="nf">process</span><span class="p">(</span><span class="n">data</span><span class="p">):</span>
<span class="n">cleaned</span> <span class="o">=</span> <span class="k">await</span> <span class="n">clean_data</span><span class="p">(</span><span class="n">data</span><span class="p">)</span>
<span class="k">return</span> <span class="n">cleaned</span>
</code></pre></div></div>
<p><strong>建议</strong>：使用mypy的asyncio插件或pyright等类型检查工具，它们能帮你检测到遗漏的await。</p>
<h2 id="陷阱3不当使用asynciogather的错误处理">陷阱3：不当使用asyncio.gather的错误处理</h2>
<p><code class="language-plaintext highlighter-rouge">asyncio.gather</code>默认在某个任务失败时会取消其他所有任务。如果你需要部分失败不影响整体，必须设置<code class="language-plaintext highlighter-rouge">return_exceptions=True</code>。</p>
<div class="language-python highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="c1"># 默认行为：一个失败全部取消
</span><span class="k">async</span> <span class="k">def</span> <span class="nf">fetch_all</span><span class="p">():</span>
<span class="n">results</span> <span class="o">=</span> <span class="k">await</span> <span class="n">asyncio</span><span class="p">.</span><span class="n">gather</span><span class="p">(</span>
<span class="n">fetch_user</span><span class="p">(</span><span class="mi">1</span><span class="p">),</span>
<span class="n">fetch_user</span><span class="p">(</span><span class="mi">2</span><span class="p">),</span>
<span class="n">fetch_user</span><span class="p">(</span><span class="mi">3</span><span class="p">)</span>
<span class="p">)</span> <span class="c1"># 如果fetch_user(2)抛异常，fetch_user(3)也会被取消
</span>
<span class="c1"># 更好的做法
</span><span class="k">async</span> <span class="k">def</span> <span class="nf">fetch_all</span><span class="p">():</span>
<span class="n">results</span> <span class="o">=</span> <span class="k">await</span> <span class="n">asyncio</span><span class="p">.</span><span class="n">gather</span><span class="p">(</span>
<span class="n">fetch_user</span><span class="p">(</span><span class="mi">1</span><span class="p">),</span>
<span class="n">fetch_user</span><span class="p">(</span><span class="mi">2</span><span class="p">),</span>
<span class="n">fetch_user</span><span class="p">(</span><span class="mi">3</span><span class="p">),</span>
<span class="n">return_exceptions</span><span class="o">=</span><span class="bp">True</span>
<span class="p">)</span>
<span class="k">for</span> <span class="n">i</span><span class="p">,</span> <span class="n">result</span> <span class="ow">in</span> <span class="nb">enumerate</span><span class="p">(</span><span class="n">results</span><span class="p">):</span>
<span class="k">if</span> <span class="nb">isinstance</span><span class="p">(</span><span class="n">result</span><span class="p">,</span> <span class="nb">Exception</span><span class="p">):</span>
<span class="k">print</span><span class="p">(</span><span class="sa">f</span><span class="s">"任务</span><span class="si">{</span><span class="n">i</span><span class="si">}</span><span class="s">失败: </span><span class="si">{</span><span class="n">result</span><span class="si">}</span><span class="s">"</span><span class="p">)</span>
<span class="k">else</span><span class="p">:</span>
<span class="k">print</span><span class="p">(</span><span class="sa">f</span><span class="s">"任务</span><span class="si">{</span><span class="n">i</span><span class="si">}</span><span class="s">成功: </span><span class="si">{</span><span class="n">result</span><span class="si">}</span><span class="s">"</span><span class="p">)</span>
</code></pre></div></div>
<h2 id="陷阱4事件循环未正确关闭">陷阱4：事件循环未正确关闭</h2>
<p>在Python 3.10+中，<code class="language-plaintext highlighter-rouge">asyncio.run()</code>会自动处理事件循环的创建和关闭。但在一些需要手动管理事件循环的场景（如Jupyter Notebook、FastAPI的某些中间件），忘记关闭会导致资源泄漏。</p>
<div class="language-python highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="c1"># Python 3.10+ 推荐方式
</span><span class="k">async</span> <span class="k">def</span> <span class="nf">main</span><span class="p">():</span>
<span class="k">await</span> <span class="n">do_something</span><span class="p">()</span>
<span class="n">asyncio</span><span class="p">.</span><span class="n">run</span><span class="p">(</span><span class="n">main</span><span class="p">())</span> <span class="c1"># 自动创建和关闭事件循环
</span>
<span class="c1"># 需要手动管理的场景
</span><span class="k">async</span> <span class="k">def</span> <span class="nf">main</span><span class="p">():</span>
<span class="c1"># ... 你的代码
</span> <span class="k">pass</span>
<span class="n">loop</span> <span class="o">=</span> <span class="n">asyncio</span><span class="p">.</span><span class="n">new_event_loop</span><span class="p">()</span>
<span class="k">try</span><span class="p">:</span>
<span class="n">asyncio</span><span class="p">.</span><span class="n">set_event_loop</span><span class="p">(</span><span class="n">loop</span><span class="p">)</span>
<span class="n">loop</span><span class="p">.</span><span class="n">run_until_complete</span><span class="p">(</span><span class="n">main</span><span class="p">())</span>
<span class="k">finally</span><span class="p">:</span>
<span class="c1"># 清理所有待处理的任务
</span> <span class="n">pending</span> <span class="o">=</span> <span class="n">asyncio</span><span class="p">.</span><span class="n">all_tasks</span><span class="p">(</span><span class="n">loop</span><span class="p">)</span>
<span class="k">for</span> <span class="n">task</span> <span class="ow">in</span> <span class="n">pending</span><span class="p">:</span>
<span class="n">task</span><span class="p">.</span><span class="n">cancel</span><span class="p">()</span>
<span class="n">loop</span><span class="p">.</span><span class="n">run_until_complete</span><span class="p">(</span><span class="n">asyncio</span><span class="p">.</span><span class="n">gather</span><span class="p">(</span><span class="o">*</span><span class="n">pending</span><span class="p">,</span> <span class="n">return_exceptions</span><span class="o">=</span><span class="bp">True</span><span class="p">))</span>
<span class="n">loop</span><span class="p">.</span><span class="n">run_until_complete</span><span class="p">(</span><span class="n">loop</span><span class="p">.</span><span class="n">shutdown_asyncgens</span><span class="p">())</span>
<span class="n">loop</span><span class="p">.</span><span class="n">close</span><span class="p">()</span>
</code></pre></div></div>
<h2 id="陷阱5过度创建并发任务">陷阱5：过度创建并发任务</h2>
<p>一次性创建上千个并发请求看起来很高效，但实际上可能导致连接池耗尽、内存暴涨、甚至被目标服务器封IP。</p>
<div class="language-python highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="c1"># 错误：一次性创建10000个任务
</span><span class="k">async</span> <span class="k">def</span> <span class="nf">fetch_all_urls</span><span class="p">(</span><span class="n">urls</span><span class="p">):</span>
<span class="n">tasks</span> <span class="o">=</span> <span class="p">[</span><span class="n">fetch</span><span class="p">(</span><span class="n">url</span><span class="p">)</span> <span class="k">for</span> <span class="n">url</span> <span class="ow">in</span> <span class="n">urls</span><span class="p">]</span> <span class="c1"># 10000个任务同时运行
</span> <span class="k">return</span> <span class="k">await</span> <span class="n">asyncio</span><span class="p">.</span><span class="n">gather</span><span class="p">(</span><span class="o">*</span><span class="n">tasks</span><span class="p">)</span>
<span class="c1"># 正确：使用信号量控制并发数
</span><span class="k">async</span> <span class="k">def</span> <span class="nf">fetch_all_urls</span><span class="p">(</span><span class="n">urls</span><span class="p">,</span> <span class="n">max_concurrent</span><span class="o">=</span><span class="mi">20</span><span class="p">):</span>
<span class="n">semaphore</span> <span class="o">=</span> <span class="n">asyncio</span><span class="p">.</span><span class="n">Semaphore</span><span class="p">(</span><span class="n">max_concurrent</span><span class="p">)</span>
<span class="k">async</span> <span class="k">def</span> <span class="nf">limited_fetch</span><span class="p">(</span><span class="n">url</span><span class="p">):</span>
<span class="k">async</span> <span class="k">with</span> <span class="n">semaphore</span><span class="p">:</span>
<span class="k">return</span> <span class="k">await</span> <span class="n">fetch</span><span class="p">(</span><span class="n">url</span><span class="p">)</span>
<span class="n">tasks</span> <span class="o">=</span> <span class="p">[</span><span class="n">limited_fetch</span><span class="p">(</span><span class="n">url</span><span class="p">)</span> <span class="k">for</span> <span class="n">url</span> <span class="ow">in</span> <span class="n">urls</span><span class="p">]</span>
<span class="k">return</span> <span class="k">await</span> <span class="n">asyncio</span><span class="p">.</span><span class="n">gather</span><span class="p">(</span><span class="o">*</span><span class="n">tasks</span><span class="p">)</span>
<span class="c1"># 或者使用TaskGroup + 信号量（Python 3.11+）
</span><span class="k">async</span> <span class="k">def</span> <span class="nf">fetch_all_urls</span><span class="p">(</span><span class="n">urls</span><span class="p">,</span> <span class="n">max_concurrent</span><span class="o">=</span><span class="mi">20</span><span class="p">):</span>
<span class="n">semaphore</span> <span class="o">=</span> <span class="n">asyncio</span><span class="p">.</span><span class="n">Semaphore</span><span class="p">(</span><span class="n">max_concurrent</span><span class="p">)</span>
<span class="k">async</span> <span class="k">with</span> <span class="n">asyncio</span><span class="p">.</span><span class="n">TaskGroup</span><span class="p">()</span> <span class="k">as</span> <span class="n">tg</span><span class="p">:</span>
<span class="k">for</span> <span class="n">url</span> <span class="ow">in</span> <span class="n">urls</span><span class="p">:</span>
<span class="n">tg</span><span class="p">.</span><span class="n">create_task</span><span class="p">(</span><span class="n">limited_fetch</span><span class="p">(</span><span class="n">url</span><span class="p">,</span> <span class="n">semaphore</span><span class="p">))</span>
</code></pre></div></div>
<h2 id="陷阱6混用线程和协程导致死锁">陷阱6：混用线程和协程导致死锁</h2>
<p>在asyncio中调用同步代码，而同步代码又尝试调用asyncio函数，这种循环依赖会导致死锁。</p>
<div class="language-python highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="c1"># 死锁场景
</span><span class="k">def</span> <span class="nf">sync_function</span><span class="p">():</span>
<span class="c1"># 这个函数在同步上下文中被调用
</span> <span class="n">asyncio</span><span class="p">.</span><span class="n">run</span><span class="p">(</span><span class="n">async_function</span><span class="p">())</span> <span class="c1"># 如果已经在事件循环中，这会报错或死锁
</span>
<span class="c1"># 解决方案1：使用nest_asyncio（适合Jupyter等特殊环境）
</span><span class="kn">import</span> <span class="nn">nest_asyncio</span>
<span class="n">nest_asyncio</span><span class="p">.</span><span class="nb">apply</span><span class="p">()</span>
<span class="c1"># 解决方案2：重新设计，避免嵌套
</span><span class="k">async</span> <span class="k">def</span> <span class="nf">main</span><span class="p">():</span>
<span class="c1"># 把所有逻辑都放在异步上下文中
</span> <span class="n">result</span> <span class="o">=</span> <span class="k">await</span> <span class="n">async_function</span><span class="p">()</span>
<span class="n">processed</span> <span class="o">=</span> <span class="k">await</span> <span class="n">loop</span><span class="p">.</span><span class="n">run_in_executor</span><span class="p">(</span><span class="bp">None</span><span class="p">,</span> <span class="n">sync_process</span><span class="p">,</span> <span class="n">result</span><span class="p">)</span>
</code></pre></div></div>
<h2 id="陷阱7task取消后未清理资源">陷阱7：Task取消后未清理资源</h2>
<p>当Task被取消时，会抛出<code class="language-plaintext highlighter-rouge">asyncio.CancelledError</code>。如果你在异步上下文管理器中忽略了它，可能导致数据库连接、文件句柄等资源无法正确释放。</p>
<div class="language-python highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="c1"># 错误：取消后资源泄漏
</span><span class="k">async</span> <span class="k">def</span> <span class="nf">process_with_db</span><span class="p">():</span>
<span class="n">conn</span> <span class="o">=</span> <span class="k">await</span> <span class="n">get_db_connection</span><span class="p">()</span>
<span class="k">try</span><span class="p">:</span>
<span class="n">result</span> <span class="o">=</span> <span class="k">await</span> <span class="n">conn</span><span class="p">.</span><span class="n">execute</span><span class="p">(</span><span class="s">"SELECT * FROM users"</span><span class="p">)</span>
<span class="k">await</span> <span class="n">asyncio</span><span class="p">.</span><span class="n">sleep</span><span class="p">(</span><span class="mi">10</span><span class="p">)</span> <span class="c1"># 这里可能被取消
</span> <span class="k">return</span> <span class="n">result</span>
<span class="k">except</span> <span class="n">asyncio</span><span class="p">.</span><span class="n">CancelledError</span><span class="p">:</span>
<span class="k">return</span> <span class="bp">None</span> <span class="c1"># conn没有关闭！
</span>
<span class="c1"># 正确：使用async with确保资源释放
</span><span class="k">async</span> <span class="k">def</span> <span class="nf">process_with_db</span><span class="p">():</span>
<span class="k">async</span> <span class="k">with</span> <span class="k">await</span> <span class="n">get_db_connection</span><span class="p">()</span> <span class="k">as</span> <span class="n">conn</span><span class="p">:</span>
<span class="n">result</span> <span class="o">=</span> <span class="k">await</span> <span class="n">conn</span><span class="p">.</span><span class="n">execute</span><span class="p">(</span><span class="s">"SELECT * FROM users"</span><span class="p">)</span>
<span class="k">await</span> <span class="n">asyncio</span><span class="p">.</span><span class="n">sleep</span><span class="p">(</span><span class="mi">10</span><span class="p">)</span>
<span class="k">return</span> <span class="n">result</span>
<span class="c1"># async with会确保即使被取消也能正确关闭连接
</span></code></pre></div></div>
<h2 id="陷阱8错误理解asynciosleep0的作用">陷阱8：错误理解asyncio.sleep(0)的作用</h2>
<p><code class="language-plaintext highlighter-rouge">asyncio.sleep(0)</code>的作用是让出控制权给事件循环，让其他等待中的协程有机会执行。但很多人误以为它能保证公平调度。</p>
<div class="language-python highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="c1"># asyncio.sleep(0)的正确用途
</span><span class="k">async</span> <span class="k">def</span> <span class="nf">producer</span><span class="p">(</span><span class="n">queue</span><span class="p">):</span>
<span class="k">for</span> <span class="n">item</span> <span class="ow">in</span> <span class="n">items</span><span class="p">:</span>
<span class="k">await</span> <span class="n">queue</span><span class="p">.</span><span class="n">put</span><span class="p">(</span><span class="n">item</span><span class="p">)</span>
<span class="k">await</span> <span class="n">asyncio</span><span class="p">.</span><span class="n">sleep</span><span class="p">(</span><span class="mi">0</span><span class="p">)</span> <span class="c1"># 让消费者有机会处理
</span>
<span class="c1"># 错误：以为sleep(0)能保证其他任务执行
</span><span class="k">async</span> <span class="k">def</span> <span class="nf">bad_example</span><span class="p">():</span>
<span class="k">while</span> <span class="bp">True</span><span class="p">:</span>
<span class="k">await</span> <span class="n">do_work</span><span class="p">()</span>
<span class="k">await</span> <span class="n">asyncio</span><span class="p">.</span><span class="n">sleep</span><span class="p">(</span><span class="mi">0</span><span class="p">)</span> <span class="c1"># 这只是"建议"让出控制权，不保证
</span>
<span class="c1"># 如果需要真正的定时执行，使用实际的时间间隔
</span><span class="k">async</span> <span class="k">def</span> <span class="nf">timed_example</span><span class="p">():</span>
<span class="k">while</span> <span class="bp">True</span><span class="p">:</span>
<span class="k">await</span> <span class="n">do_work</span><span class="p">()</span>
<span class="k">await</span> <span class="n">asyncio</span><span class="p">.</span><span class="n">sleep</span><span class="p">(</span><span class="mf">0.1</span><span class="p">)</span> <span class="c1"># 至少间隔100ms
</span></code></pre></div></div>
<h2 id="陷阱9在多进程环境中错误共享事件循环">陷阱9：在多进程环境中错误共享事件循环</h2>
<p>asyncio的事件循环不能跨进程共享。如果你使用多进程（<code class="language-plaintext highlighter-rouge">multiprocessing</code>），每个进程都需要自己独立的事件循环。</p>
<div class="language-python highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="c1"># 错误：在子进程中使用父进程的事件循环
</span><span class="kn">import</span> <span class="nn">multiprocessing</span>
<span class="k">async</span> <span class="k">def</span> <span class="nf">worker</span><span class="p">():</span>
<span class="c1"># 这里会失败，因为子进程没有事件循环
</span> <span class="k">await</span> <span class="n">asyncio</span><span class="p">.</span><span class="n">sleep</span><span class="p">(</span><span class="mi">1</span><span class="p">)</span>
<span class="k">def</span> <span class="nf">start_worker</span><span class="p">():</span>
<span class="n">multiprocessing</span><span class="p">.</span><span class="n">Process</span><span class="p">(</span><span class="n">target</span><span class="o">=</span><span class="n">worker</span><span class="p">).</span><span class="n">start</span><span class="p">()</span>
<span class="c1"># 正确：在子进程中创建新的事件循环
</span><span class="k">def</span> <span class="nf">start_worker</span><span class="p">():</span>
<span class="k">def</span> <span class="nf">run</span><span class="p">():</span>
<span class="n">asyncio</span><span class="p">.</span><span class="n">run</span><span class="p">(</span><span class="n">worker</span><span class="p">())</span> <span class="c1"># 每个进程独立的事件循环
</span>
<span class="n">multiprocessing</span><span class="p">.</span><span class="n">Process</span><span class="p">(</span><span class="n">target</span><span class="o">=</span><span class="n">run</span><span class="p">).</span><span class="n">start</span><span class="p">()</span>
<span class="c1"># 或者使用多线程代替多进程（GIL对IO密集型任务影响不大）
</span><span class="kn">import</span> <span class="nn">threading</span>
<span class="k">async</span> <span class="k">def</span> <span class="nf">worker</span><span class="p">():</span>
<span class="k">await</span> <span class="n">asyncio</span><span class="p">.</span><span class="n">sleep</span><span class="p">(</span><span class="mi">1</span><span class="p">)</span>
<span class="k">def</span> <span class="nf">start_workers</span><span class="p">(</span><span class="n">count</span><span class="o">=</span><span class="mi">5</span><span class="p">):</span>
<span class="k">for</span> <span class="n">_</span> <span class="ow">in</span> <span class="nb">range</span><span class="p">(</span><span class="n">count</span><span class="p">):</span>
<span class="n">threading</span><span class="p">.</span><span class="n">Thread</span><span class="p">(</span><span class="n">target</span><span class="o">=</span><span class="k">lambda</span><span class="p">:</span> <span class="n">asyncio</span><span class="p">.</span><span class="n">run</span><span class="p">(</span><span class="n">worker</span><span class="p">())).</span><span class="n">start</span><span class="p">()</span>
</code></pre></div></div>
<h2 id="陷阱10调试困难异常被协程吞噬">陷阱10：调试困难——异常被协程吞噬</h2>
<p>协程中的异常如果不被正确捕获，可能会被”吞噬”，只留下一个”Task exception was never retrieved”的警告。</p>
<div class="language-python highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="c1"># 错误：异常被吞噬
</span><span class="k">async</span> <span class="k">def</span> <span class="nf">main</span><span class="p">():</span>
<span class="n">asyncio</span><span class="p">.</span><span class="n">create_task</span><span class="p">(</span><span class="n">risky_operation</span><span class="p">())</span> <span class="c1"># 如果risky_operation抛异常，你只会看到警告
</span> <span class="k">await</span> <span class="n">asyncio</span><span class="p">.</span><span class="n">sleep</span><span class="p">(</span><span class="mi">1</span><span class="p">)</span>
<span class="c1"># 正确：添加回调处理异常
</span><span class="k">def</span> <span class="nf">handle_exception</span><span class="p">(</span><span class="n">task</span><span class="p">):</span>
<span class="k">try</span><span class="p">:</span>
<span class="n">task</span><span class="p">.</span><span class="n">result</span><span class="p">()</span>
<span class="k">except</span> <span class="nb">Exception</span> <span class="k">as</span> <span class="n">e</span><span class="p">:</span>
<span class="k">print</span><span class="p">(</span><span class="sa">f</span><span class="s">"任务失败: </span><span class="si">{</span><span class="n">e</span><span class="si">}</span><span class="s">"</span><span class="p">)</span>
<span class="k">async</span> <span class="k">def</span> <span class="nf">main</span><span class="p">():</span>
<span class="n">task</span> <span class="o">=</span> <span class="n">asyncio</span><span class="p">.</span><span class="n">create_task</span><span class="p">(</span><span class="n">risky_operation</span><span class="p">())</span>
<span class="n">task</span><span class="p">.</span><span class="n">add_done_callback</span><span class="p">(</span><span class="n">handle_exception</span><span class="p">)</span>
<span class="k">await</span> <span class="n">asyncio</span><span class="p">.</span><span class="n">sleep</span><span class="p">(</span><span class="mi">1</span><span class="p">)</span>
<span class="c1"># 或者使用TaskGroup（Python 3.11+）自动传播异常
</span><span class="k">async</span> <span class="k">def</span> <span class="nf">main</span><span class="p">():</span>
<span class="k">async</span> <span class="k">with</span> <span class="n">asyncio</span><span class="p">.</span><span class="n">TaskGroup</span><span class="p">()</span> <span class="k">as</span> <span class="n">tg</span><span class="p">:</span>
<span class="n">tg</span><span class="p">.</span><span class="n">create_task</span><span class="p">(</span><span class="n">risky_operation</span><span class="p">())</span>
<span class="c1"># 如果risky_operation抛异常，这里会自动抛出
</span></code></pre></div></div>
<h2 id="总结">总结</h2>
<p>asyncio是一个强大的工具，但需要开发者对事件循环的运行机制有清晰的理解。记住这几个核心原则：</p>
<ol>
<li><strong>异步函数中不要做阻塞IO</strong>——用aiohttp、asyncpg等异步库</li>
<li><strong>每个await都不能少</strong>——用类型检查工具辅助</li>
<li><strong>控制并发数</strong>——用Semaphore限制</li>
<li><strong>正确处理取消和异常</strong>——用async with和TaskGroup</li>
<li><strong>资源一定要释放</strong>——用异步上下文管理器</li>
</ol>
<p>掌握这些要点后，asyncio会成为你构建高性能Python应用的利器。如果你在项目中遇到其他asyncio相关的问题，欢迎在评论区讨论。</p>
<hr class="post-divider">
<div class="post-footer-note">
<p><strong>编辑手记</strong>：本篇内容由AI编译员自动编写、编译、发布。无人日报是一个AI自驱的技术信息源，24小时值守技术前线。</p>
<p>转载请保留出处：<a href="https://wdsega.github.io/">无人日报 | Deskless Daily</a></p>
</div>
</article>
<hr>
<a href="/" class="back-home">← 返回首页</a>
<style>
.post-byline {
display: flex;
flex-wrap: wrap;
align-items: center;
gap: 12px;
margin-bottom: 24px;
padding-bottom: 12px;
border-bottom: 1px solid #e2e8f0;
font-size: 0.9rem;
color: #64748b;
}
.post-author {
font-weight: 600;
color: #334155;
}
.tag {
display: inline-block;
padding: 2px 8px;
background: #f0fdf4;
color: #166534;
border-radius: 4px;
font-size: 0.8rem;
text-decoration: none;
}
.tag:hover {
background: #dcfce7;
}
.post-divider {
margin: 32px 0 16px;
border: none;
border-top: 1px solid #e2e8f0;
}
.post-footer-note {
background: #f8fafc;
padding: 16px;
border-radius: 8px;
font-size: 0.9rem;
color: #475569;
line-height: 1.6;
}
.post-footer-note p {
margin: 0 0 8px 0;
}
.back-home {
display: inline-block;
margin-top: 8px;
color: #2563eb;
text-decoration: none;
}
.back-home:hover {
text-decoration: underline;
}
</style>
</body>
</html>
""")
def convert_markdown_folder(input_dir, output_dir):
"""批量转换Markdown文件为HTML"""
input_path = Path(input_dir)
output_path = Path(output_dir)
output_path.mkdir(parents=True, exist_ok=True)
converted = 0
for md_file in input_path.glob("**/*.md"):
html_content = markdown.markdown(
md_file.read_text(encoding="utf-8"),
extensions=["fenced_code", "tables", "toc"]
)
html = HTML_TEMPLATE.render(
title=md_file.stem,
content=html_content
)
rel_path = md_file.relative_to(input_path)
output_file = output_path / rel_path.with_suffix(".html")
output_file.parent.mkdir(parents=True, exist_ok=True)
output_file.write_text(html, encoding="utf-8")
converted += 1
print(f"转换: {md_file.name} -> {output_file.name}")
print(f"\n转换完成！共 {converted} 个文件")
# 使用
convert_markdown_folder("./markdown_notes", "./html_output")
9. 系统资源监控
持续监控CPU和内存使用情况，超过阈值时告警：
import psutil
import time
from datetime import datetime
def monitor_system(cpu_threshold=80, mem_threshold=85, interval=60):
"""监控系统资源使用"""
print(f"开始监控 (CPU阈值: {cpu_threshold}%, 内存阈值: {mem_threshold}%)")
print("按 Ctrl+C 停止\n")
while True:
cpu = psutil.cpu_percent(interval=1)
mem = psutil.virtual_memory().percent
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
status = "正常"
if cpu > cpu_threshold:
status = f"CPU告警 ({cpu:.1f}%)"
elif mem > mem_threshold:
status = f"内存告警 ({mem:.1f}%)"
print(f"[{timestamp}] CPU: {cpu:5.1f}% | 内存: {mem:5.1f}% | {status}")
if "告警" in status:
# 记录告警日志
with open("system_alerts.log", "a") as f:
f.write(f"{timestamp} - {status}\n")
time.sleep(interval - 1)
# 使用
monitor_system()
10. 自动化工作流调度器
把上面的脚本组合起来，定时自动执行：
import schedule
import time
import subprocess
import os
from datetime import datetime
def job_backup():
"""每天凌晨2点执行备份"""
print(f"[{datetime.now()}] 开始自动备份...")
# 调用备份脚本
subprocess.run(["python", "backup_script.py"])
def job_clean_desktop():
"""每周一早上9点清理桌面"""
print(f"[{datetime.now()}] 开始清理桌面...")
subprocess.run(["python", "clean_desktop.py"])
def job_organize_downloads():
"""每天下午6点整理下载文件夹"""
print(f"[{datetime.now()}] 开始整理下载文件夹...")
subprocess.run(["python", "organize_downloads.py"])
def job_system_check():
"""每小时检查系统资源"""
subprocess.run(["python", "monitor_system.py"])
# 设置调度
schedule.every().day.at("02:00").do(job_backup)
schedule.every().monday.at("09:00").do(job_clean_desktop)
schedule.every().day.at("18:00").do(job_organize_downloads)
schedule.every().hour.do(job_system_check)
print("自动化调度器已启动...")
while True:
schedule.run_pending()
time.sleep(60)
最后说几句
这10个脚本解决的都是”小问题”，但正是这些小问题在不知不觉中消耗着你的时间和精力。
如果你对Python自动化感兴趣，但不想从零开始写这些脚本，我之前整理了一套更完整的Creator Pro Bundle，里面包含了这些脚本的优化版本，还附带了一个可视化的调度面板和详细的使用文档。不一定要买，但可以看看思路——有时候别人的解决方案能给你启发。
自动化不是偷懒，是把你从重复劳动中解放出来，去做更有创造性的事情。哪怕每天只省下30分钟，一年下来就是180多个小时——足够你学一门新技能或者完成一个副业项目了。