TinyPNG 的免费 API 每个账号每月 500 张。要处理更多图片，最直接的办法就是注册多个 Key。

但多 Key 带来了一个新问题：多个线程同时拿 Key，怎么保证不冲突？

场景	问题
两个线程拿到同一个 Key	API 调用互相干扰，配额混乱
一个线程用着 Key 被另一个线程释放	状态不同步，Key 被重复使用
Key 额度用完了还在继续用	浪费请求，全部返回 429
线程数超过 Key 数	部分线程永远拿不到 Key，死等

这篇文章拆解 TinyOpt 中 KeyManager 的完整实现——一个生产环境验证过的线程安全 API Key 池。

01. 需求分析：我们要什么

一个 Key 管理模块需要满足 5 个要求：

#	要求	说明
1	线程安全	N 个 Worker 同时获取 Key，不冲突
2	公平轮询	所有 Key 被均匀使用，避免单个 Key 过早耗尽
3	自动禁用	Key 额度用完或失效时自动跳过，不浪费请求
4	防重复获取	一个 Key 同时只能被一个 Worker 持有
5	无锁竞争	高并发下获取 Key 的操作不能成为瓶颈

02. 数据结构设计

每个 Key 用一个对象表示：

@dataclass
class APIKey:
    key: str
    month_quota: int = 500
    used_count: int = 0
    disabled: bool = False
    in_use: bool = False

KeyManager 维护一个 Key 列表和一个 threading.Lock：

class KeyManager:
    def __init__(self):
        self.keys: list[APIKey] = []
        self._lock = threading.Lock()
        self._index = 0

设计思考：

用 Lock 而非 RLock——获取 Key 的操作是互斥的，不需要重入
in_use 标志防止同一个 Key 被两个 Worker 同时持有
disabled 标志让配额耗尽的 Key 永久跳过

03. 核心方法：acquire_key

获取 Key 是整个模块最核心的方法。它需要在线程安全的前提下，从可用 Key 中选出一个最合适的：

def acquire_key(self) -> APIKey | None:
    with self._lock:
        n = len(self.keys)
        if n == 0:
            return None

        for i in range(n):
            idx = (self._index + i) % n
            key = self.keys[idx]
            if not key.in_use and not key.disabled:
                key.in_use = True
                self._index = (idx + 1) % n
                return key
        return None

关键细节：

① 遍历全部 Key 而非只查下一个 从 _index 开始遍历一圈，而不是只看下一个。这保证了：即使当前指针指向一个正在使用中的 Key，也能找到可用的 Key。

② 更新指针实现公平轮询 每次分配后指针后移一位，下次分配从下一个位置开始。所有 Key 被均匀使用，不会出现前几个 Key 用完、后面的 Key 空闲的情况。

③ 返回 None 而非阻塞 没有可用 Key 时直接返回 None，由调用方决定等待、重试或降级。这避免了在线程池中产生死锁。

release_key 是对称的：

def release_key(self, key: APIKey):
    with self._lock:
        key.in_use = False

04. 配额耗尽自动禁用

每次压缩完成后，检查 API 返回的剩余配额：

def process_one_file(self, file_path):
    key = self.key_manager.acquire_key()
    if key is None:
        self.signal_progress.emit(file_path, "waiting")
        return False

    try:
        response = self._call_tinypng(file_path, key.key)
        remaining = response.headers.get("Compression-Count")
        if remaining and int(remaining) <= 0:
            key.disabled = True
            self.signal_log.emit(f"Key {key.key[:8]}... 额度已用完，已自动禁用")
        key.used_count += 1
        return True
    except TinyPNGQuotaExceeded:
        key.disabled = True
        return False
    finally:
        self.key_manager.release_key(key)

用了 try/finally 确保无论成功还是失败，Key 都会被释放。

05. 并发配置

KeyManager 本身不限制并发数。并发由 CompressWorker 控制：

max_workers = min(len(key_manager.keys), 3, len(files))

三个限制取最小值：

可用的 Key 数量（每个 Key 同时只能处理一张）
硬限制 3（TinyPNG API 的隐形并发上限）
待处理文件数量

06. 性能对比

配置	500 张图片耗时	API 调用成功率	Key 配额利用率
单 Key 单线程	~15 min	100%	100%
3 Key 单线程	~45 min（轮询）	100%	100%
3 Key 3 并发（无锁）	~6 min	62%（冲突导致）	~73%
3 Key 3 并发（KeyManager）	~5 min	100%	100%

无锁版本 62% 的成功率和 73% 的配额利用率说明了一切——多个线程不加控制地拿 Key，不仅浪费请求，还浪费额度。

KeyManager 版本：

500 张图片从 15 分钟压缩到 5 分钟
3 个 Key 的配额被完整利用
零配额浪费，零冲突

07. 这个模式能用在别处吗

能。KeyManager 本质上是一个资源池模式。凡是有限资源、多个消费者并发访问的场景，都可以复用：

场景	资源	消费者
API Key 池	第三方 API 密钥	并发请求 Worker
数据库连接池	数据库连接	查询线程
代理 IP 池	代理服务器	爬虫 Worker
限速令牌桶	请求配额	API 调用者

核心思想是一样的：用锁保护状态变更，用轮询保证公平，用标志位追踪资源状态。

08. 总结

写完 KeyManager 之后，最深的感受是：并发编程的难点不在于锁，而在于没有意识到需要锁。

当 KeyManager 只有 50 行时加锁非常简单。当你意识到需要锁的时候，Key 已经在 3 个类里被传递、5 个地方被修改了——这时候再加锁，已经很难保证覆盖全了。

建议	说明
尽早抽象	第一个 Key 的时候就用 KeyManager，不要等第 5 个
缩小锁范围	只锁状态判断和更新，不锁 API 调用
try/finally	确保所有路径都释放 Key，包括异常路径
不阻塞	没资源时返回 None，让上层决策

👉 TinyOpt 开箱即用，内置多 Key 管理。如果你正在做类似的资源池设计，源码中的 KeyManager 可以直接参考。

手写一个线程安全的 API Key 池：Python 并发编程最佳实践