图文博客里，图片往往比正文更占空间。一张手机照片直接传上去，页面上只显示几百像素宽，浏览器却要下载整张原图。文章多起来以后，备份包越来越大，换主题、迁移站点时，还得跟着搬一遍。
逐张打开图片、改尺寸、另存为，处理几张还行。面对一个按年份和文章分类的图片目录，我更愿意把规则写进脚本：原图留着，批量生成适合网页展示的版本，同时记下处理结果。
这篇文章围绕这个任务展开。计算放在阿里云 ECS 上，Python 负责遍历目录和分配任务，Pillow 完成缩放与 WebP 编码。重点放在三个问题上：图片该怎么处理，多核算力怎么用，以及处理完之后怎么判断这次优化值不值。
1. 先把处理规则定下来
这里处理的是博客配图中的静态照片，输入格式为 JPEG 和 PNG。规则很简单：
- 长边超过 1920 像素时等比例缩小，小图保留原尺寸。
- 输出 WebP，使用有损编码，
quality=82、method=4。 - 保留目录层级，原文件名后追加
.webp。 原图单独保存，每次任务写入新的输出目录。
1920和82是这套方案的起始参数。最终取值要拿实际页面看：照片边缘、暗部和渐变有没有明显损伤，放大查看是否还够用。
文字密集的终端截图、二维码、架构图放在另一个目录，继续使用 PNG。照片和截图的关注点不同，给所有图片套同一组参数，后面很容易花时间返工。
WebP 的quality控制有损编码质量；method在编码速度与压缩效果之间取舍，范围为 0—6。这两个参数的含义可以直接查 Pillow WebP 文档。
整个流程如下：本地照片目录 ↓ 上传 ECS 工作目录 → 枚举图片 → 进程池处理 → WebP 目录 ├── manifest.jsonl └── summary.json
2. ECS 环境：先记配置，再谈性能
下面的部署命令以 Ubuntu 24.04 为例。实例可以从 u2a 的 4 vCPU、8 GiB 规格起步，对应 ecs.u2a-c1m2.xlarge。这给 1、2、4、8 个进程的对照留出了观察空间，8 个进程用来观察超过 vCPU 数量后的变化。
这里有个选型细节：阿里云文档写明，u2a 兼容多代 AMD EPYC 处理器，支持 AMD Turin。因此，实际 CPU 型号也要记录，后面的性能分析才有依据。规格信息见 阿里云 u2a 官方文档。
创建实例后，通过 SSH 登录。这个任务通过命令行运行，安全组的 SSH 入方向来源限定为自己的公网 IP 即可。
安装环境：
sudo apt-get update
sudo apt-get install -y python3-venv sysstat
mkdir -p ~/image-job/raw ~/image-job/results
cd ~/image-job
python3 -m venv .venv
source .venv/bin/activate
python -m pip install Pillow==12.3.0
lscpu > cpu.txt
uname -a > system.txt
python --version > python-version.txt
python -m pip freeze > requirements.txt
python -c "from PIL import features; print('WebP:', features.check('webp'))"
最后一条命令应输出 WebP: True。把照片上传到 raw/，把下面的脚本保存为 batch_webp.py，就可以开始处理。
3. 脚本里几个容易忽略的细节
单张图片的处理顺序是：打开文件、纠正方向、转换颜色模式、缩放、编码。
手机照片的显示方向可能写在 EXIF 里。如果只读取像素再另存，输出图有可能横过来。脚本先调用 ImageOps.exif_transpose()，把方向落实到像素上。这一步的行为见 Pillow ImageOps 文档。
透明图片也要单独判断。带 Alpha 通道的图，以及使用调色板透明信息的 PNG，都转成 RGBA；其他图片转成 RGB。源图已有的 ICC 配置随输出保存。
文件名用 photo.jpg.webp 这样的形式。有些目录里会同时存在 photo.jpg 和 photo.png，如果都写成 photo.webp，就会发生覆盖。多留一个原始后缀，回查来源也方便。
完整脚本如下：
import argparse
import json
import time
from concurrent.futures import ProcessPoolExecutor
from pathlib import Path
from PIL import Image, ImageOps
def convert_one(job):
source, target = map(Path, job)
try:
target.parent.mkdir(parents=True, exist_ok=True)
with Image.open(source) as original:
if getattr(original, "n_frames", 1) != 1:
raise ValueError("Only static images are supported")
image = ImageOps.exif_transpose(original)
icc = image.info.get("icc_profile", b"")
has_alpha = "A" in image.getbands() or "transparency" in image.info
image = image.convert("RGBA" if has_alpha else "RGB")
image.thumbnail((1920, 1920), Image.Resampling.LANCZOS)
image.save(target, "WEBP", quality=82, method=4, icc_profile=icc)
return {
"source": str(source), "target": str(target), "status": "ok",
"input_bytes": source.stat().st_size,
"output_bytes": target.stat().st_size,
}
except (OSError, ValueError) as exc:
return {
"source": str(source), "status": "error", "error": str(exc)}
def main():
parser = argparse.ArgumentParser()
parser.add_argument("source", type=Path)
parser.add_argument("target", type=Path)
parser.add_argument("--workers", type=int, default=4)
args = parser.parse_args()
if args.workers < 1:
parser.error("--workers must be positive")
files = sorted(
p for p in args.source.rglob("*")
if p.is_file() and p.suffix.lower() in {
".jpg", ".jpeg", ".png"}
)
if not files:
parser.error("No JPEG or PNG files found")
args.target.mkdir(parents=True, exist_ok=False)
jobs = [
(str(p), str(args.target / p.relative_to(args.source).with_name(p.name + ".webp")))
for p in files
]
started = time.perf_counter()
with ProcessPoolExecutor(max_workers=args.workers) as pool:
rows = list(pool.map(convert_one, jobs, chunksize=1))
seconds = time.perf_counter() - started
with (args.target / "manifest.jsonl").open("w", encoding="utf-8") as stream:
for row in rows:
stream.write(json.dumps(row, ensure_ascii=False) + "\n")
good = [r for r in rows if r["status"] == "ok"]
input_bytes = sum(r["input_bytes"] for r in good)
output_bytes = sum(r["output_bytes"] for r in good)
summary = {
"workers": args.workers, "success": len(good),
"failed": len(rows) - len(good), "seconds": round(seconds, 3),
"images_per_second": round(len(good) / seconds, 3),
"input_bytes": input_bytes, "output_bytes": output_bytes,
"saved_percent": round(100 * (1 - output_bytes / input_bytes), 2)
if input_bytes else None,
}
(args.target / "summary.json").write_text(
json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8"
)
print(json.dumps(summary, ensure_ascii=False, indent=2))
raise SystemExit(1 if summary["failed"] else 0)
if __name__ == "__main__":
main()
执行一次：
python batch_webp.py raw results/first-run --workers 4
输出目录里，manifest.jsonl 逐张记录源文件、目标文件、状态和体积；summary.json 汇总成功数、失败数、耗时和总大小。坏文件和动态 PNG 会留下错误记录，其他图片继续处理。
主进程把路径交给子进程，子进程自己打开图片。这样进程间传递的是文件路径，解码后的大块像素数据留在各自的进程里。这里使用 Python 标准库的 ProcessPoolExecutor，后续通过实测决定进程数。
4. 并发开多少，用同一批图片比较
处理规则确定以后，再调整并发数。否则，一边降低图片质量，一边增加进程，最后很难解释耗时为什么变了。
先完整运行一次，让依赖和输入数据经过首次读取。预热结果放在独立目录：
python batch_webp.py raw results/warmup --workers 1
接着分别测试 1、2、4、8 个进程，每组跑三次：
for run in 1 2 3; do
for workers in 1 2 4 8; do
python batch_webp.py raw "results/r${run}-w${workers}" \
--workers "$workers"
done
done
每轮使用新目录，编码参数和输入文件保持一致。比较前先检查各组成功数和失败数，确认处理的是同一批文件。数据接近时，可以再轮换各组的执行顺序，观察排序是否影响结论。
脚本统计的是处理阶段的墙钟时间，包括进程池启动、图片读取、缩放、编码和输出写入；上传、下载、环境安装都在计时区间之外。这次比较的是预热后的批处理表现。
运行期间，在另一个终端观察：
mpstat -P ALL 1
iostat -xz 1
free -h
ECS 控制台也可以查看 CPU 等基础监控。需要内存等操作系统级指标时，可以安装云监控插件，具体采集范围见 阿里云实例运维说明。短任务的变化用终端观察更直接。
我会先看吞吐，再看 CPU、内存和磁盘指标。如果进程从 4 个增加到 8 个，吞吐已经不变，内存占用却继续增长，就把进程数收回到 4。假如 CPU 还有空闲，磁盘等待持续升高，则继续检查输入输出阶段。
图片解码后的内存占用也要算进去。一张 4000×3000 的 RGBA 图片，仅像素数据就约 48 MB，缩放和编码过程还需要额外空间。进程数越多，同时驻留的图片也越多。
5. 这套方案适合用到哪里
这套处理方式适合集中整理照片、迁移博客，以及批量生成网页配图。输入目录和编码参数固定后，换一批照片仍然可以沿用同一个入口，失败文件也有清单可查。
ECS 在这里承担的是一个独立的批处理环境：运行版本固定，任务可以留在远端执行，也便于在同一套代码下比较不同实例规格。实际使用时，端到端收益还要把素材上传和结果下载的时间算进去。
还有一个容易漏掉的收尾工作：脚本保留了原文件后缀，博客引用也要相应改为 photo.jpg.webp，并检查页面是否正常加载。先挑一篇文章替换，确认路径和显示效果，再批量修改。
我更看重这项工作的可重复性。压缩规则、运行环境和结果记录齐了，下一次就能直接回答：这批图片用了多久、缩小了多少、哪些文件需要重新处理。后续是否换更大的实例，也有数据可比较。
