多进程环境下用Cachetools缓存TClient减少API调用的可行性咨询

• 直接用cachetools.TTLCache装饰函数无法实现多进程共享：每个Python进程有独立内存空间，各自的缓存实例互不干扰，达不到共享缓存、减少API调用的目的。
• 结合进程间通信（IPC）+独立缓存进程是可行的：让所有业务进程通过队列/管道从同一个缓存管理进程获取TClient实例，统一控制令牌刷新频率，既能符合DRY原则，又能将API调用限制为每小时一次。

1. 缓存管理模块（cache_manager.py） #

实现独立的缓存进程，用TTLCache维护TClient，处理其他进程的获取请求：

from cachetools import TTLCache
from multiprocessing import Process, Queue
import time

class TClient:
    def __init__(self):
        # 初始化时触发API请求刷新令牌
        self.token = self._refresh_token()
        print(f"[TClient] 刷新令牌: {self.token}")

    def _refresh_token(self):
        # 实际逻辑：调用API获取新令牌
        return f"token_{int(time.time()//3600)}"  # 每小时生成唯一令牌标识

    def api_method(self):
        return f"使用令牌 {self.token} 调用API"

def cache_worker(request_queue, response_queue):
    # 初始化TTL缓存：最大容量1，过期时间3600秒（1小时）
    cache = TTLCache(maxsize=1, ttl=3600)
    cache_key = "shared_tclient"

    while True:
        if not request_queue.empty():
            req = request_queue.get()
            if req == "GET_CLIENT":
                # 缓存过期/不存在时，重新创建TClient实例
                if cache_key not in cache:
                    cache[cache_key] = TClient()
                # 返回缓存的实例给业务进程
                response_queue.put(cache[cache_key])
            elif req == "EXIT":
                break
        time.sleep(0.1)

2. 业务进程模块（以process1.py为例） #

业务进程通过队列向缓存管理进程请求TClient：

from multiprocessing import Queue

def run(request_queue, response_queue):
    # 发送获取客户端的请求
    request_queue.put("GET_CLIENT")
    client = response_queue.get()

    # 业务逻辑：使用客户端调用API
    print(f"[业务进程] {client.api_method()}")
    # 后续业务代码...

3. 主进程（main.py） #

启动缓存管理进程和4个业务进程：

from multiprocessing import Process, Queue
import process1  # 同理导入process2、process3、process4

if __name__ == "__main__":
    # 创建进程间通信队列
    request_queue = Queue()
    response_queue = Queue()

    # 启动缓存管理进程
    cache_process = Process(target=cache_worker, args=(request_queue, response_queue))
    cache_process.start()

    # 启动4个业务进程
    business_processes = []
    for _ in range(4):
        p = Process(target=process1.run, args=(request_queue, response_queue))
        business_processes.append(p)
        p.start()

    # 等待所有业务进程执行完成
    for p in business_processes:
        p.join()

    # 停止缓存管理进程
    request_queue.put("EXIT")
    cache_process.join()

• TClient必须可序列化：通过队列传递对象时会触发pickle序列化，确保类的所有属性都能被正常序列化，避免报错。
• 异常处理补充：实际场景中要给缓存进程添加API调用失败的重试逻辑，防止缓存进程崩溃导致业务进程无法获取客户端。
• 主动刷新可选：如果不想依赖TTL过期时的被动刷新，可以在缓存进程中添加定时任务，提前刷新令牌。
• 并发安全：缓存进程是单进程运行，TTLCache本身是线程安全的，无需额外加锁处理并发请求。

内容的提问来源于stack exchange，提问作者Godzilla74