4. I/O优化：文件读写优化、网络请求优化、数据库查询优化
聊到自动化脚本的性能，I/O这块绝对是绕不开的坎儿。我见过太多脚本，CPU利用率不到10%，但跑起来慢得像蜗牛。为什么？因为大部分时间都花在等I/O上了。说白了，你的脚本不是在「算」，而是在「等」。
我个人习惯把I/O优化分成三大块：文件读写、网络请求、数据库查询。咱们一个一个来拆解。
4.1 文件读写优化
文件读写，听起来简单吧？但坑是真不少。我记得有一次帮团队优化一个日志分析脚本，原本要跑40分钟，优化后只要3分钟。你猜怎么着？就改了一行代码。
4.1.1 使用缓冲区
很多人写文件喜欢一行一行写：
# 不推荐：逐行写入，频繁I/O
with open('output.txt', 'w') as f:
for item in data:
f.write(item + '\n')
这样写，每写一行就要做一次系统调用。数据量大了，性能直接崩掉。
我建议改成批量写入：
# 推荐：批量写入，减少系统调用
with open('output.txt', 'w', buffering=65536) as f: # 64KB缓冲区
f.write('\n'.join(data))
这里的关键是 buffering 参数。默认是8KB，你可以根据文件大小调整。我一般设到64KB或128KB，效果不错。
csv.writer 或 json.dump，它们内部已经做了缓冲优化。别自己造轮子。
4.1.2 大文件分块读取
处理大文件时，千万别一次性读到内存里。我曾经见过一个同事，用 readlines() 读一个5GB的日志文件，结果服务器直接OOM了。
正确的做法是分块读取：
# 推荐：分块读取，控制内存
def process_large_file(filepath, chunk_size=8192):
with open(filepath, 'r') as f:
while True:
chunk = f.read(chunk_size)
if not chunk:
break
# 处理chunk
process(chunk)
或者用迭代器方式：
# 更Pythonic的方式
with open('large.log', 'r') as f:
for line in f: # 内部自动缓冲
process(line)
嗯，这里要注意：for line in f 这种方式虽然方便，但如果你需要随机访问文件的不同位置，还是用 seek() 配合分块读取更灵活。
4.2 网络请求优化
网络I/O是另一个大头。自动化脚本经常要调API、拉数据，如果处理不好，一个请求等几秒，几百个请求下来，时间全浪费了。
4.2.1 连接池
你想想看，每次发HTTP请求都新建一个TCP连接，三次握手、四次挥手，光这个开销就够受的。连接池就是解决这个问题的。
我用 requests.Session 配合 urllib3 的连接池：
import requests
from requests.adapters import HTTPAdapter
session = requests.Session()
adapter = HTTPAdapter(
pool_connections=20, # 连接池大小
pool_maxsize=100, # 最大连接数
max_retries=3, # 重试次数
pool_block=False # 池满时是否阻塞
)
session.mount('http://', adapter)
session.mount('https://', adapter)
# 复用session发送请求
for url in url_list:
response = session.get(url, timeout=5)
# 处理响应
4.2.2 批量操作
单个请求效率低，那就批量发。但批量不是简单地把请求堆在一起，要考虑并发控制。
我常用的方案是用 concurrent.futures 配合信号量：
from concurrent.futures import ThreadPoolExecutor, as_completed
import threading
semaphore = threading.Semaphore(10) # 控制并发数
def fetch_url(url):
with semaphore:
return session.get(url, timeout=5)
with ThreadPoolExecutor(max_workers=20) as executor:
futures = [executor.submit(fetch_url, url) for url in url_list]
for future in as_completed(futures):
result = future.result()
# 处理结果
为什么用信号量？因为 ThreadPoolExecutor 的 max_workers 控制的是线程数，但每个线程可能同时发多个请求（比如用了异步）。信号量能精确控制并发请求数。
4.3 数据库查询优化
数据库I/O，说白了就是「查」和「写」的优化。我见过太多脚本，数据库查询慢不是因为数据量大，而是因为SQL写得烂。
4.3.1 索引优化
索引是数据库查询加速的利器。但索引不是越多越好，也不是随便建个索引就完事了。
我总结了几条经验：
- 覆盖索引： 查询的字段都在索引里，避免回表查询
- 复合索引： 把查询条件中最常用的字段放前面
- 避免函数操作： 不要在索引字段上使用函数，比如
WHERE DATE(create_time) = '2024-01-01'会让索引失效
举个例子：
-- 不推荐：索引失效
SELECT * FROM orders WHERE YEAR(create_time) = 2024;
-- 推荐：利用索引
SELECT * FROM orders
WHERE create_time >= '2024-01-01'
AND create_time < '2025-01-01';
我记得有一次，一个报表脚本跑了一个小时。我一看，查询条件里用了 LIKE '%keyword%'，全表扫描。改成全文索引后，10秒就出结果了。
4.3.2 批量提交
批量插入是数据库写入优化的核心。逐条插入不仅慢，还会产生大量事务日志。
我一般这样写：
import sqlite3
conn = sqlite3.connect('data.db')
cursor = conn.cursor()
# 批量插入
data = [(1, 'Alice'), (2, 'Bob'), (3, 'Charlie')]
cursor.executemany(
'INSERT INTO users (id, name) VALUES (?, ?)',
data
)
conn.commit() # 一次性提交
对于MySQL，还可以用 LOAD DATA INFILE 或者批量INSERT语句：
# 批量INSERT
INSERT INTO users (id, name) VALUES
(1, 'Alice'),
(2, 'Bob'),
(3, 'Charlie');
4.3.3 连接池与事务管理
数据库连接池和HTTP连接池原理类似。我用 DBUtils 或 SQLAlchemy 的连接池：
from dbutils.pooled_db import PooledDB
import pymysql
pool = PooledDB(
creator=pymysql,
maxconnections=10, # 最大连接数
mincached=2, # 最小空闲连接
maxcached=5, # 最大空闲连接
blocking=True, # 无连接时是否阻塞
host='localhost',
user='root',
password='password',
database='test'
)
# 从池中获取连接
conn = pool.connection()
cursor = conn.cursor()
cursor.execute('SELECT * FROM users')
conn.close() # 不是真的关闭，而是归还到连接池
事务管理也很重要。我习惯用上下文管理器：
with conn.cursor() as cursor:
conn.begin() # 开启事务
try:
cursor.execute('UPDATE accounts SET balance = balance - 100 WHERE id = 1')
cursor.execute('UPDATE accounts SET balance = balance + 100 WHERE id = 2')
conn.commit()
except Exception:
conn.rollback()
raise
好了，这一章的内容就这些。下一章我们会聊「内存管理与垃圾回收」，到时候再细说Python的内存模型和优化技巧。