异步定时调用多API执行固定流程可采用的设计模式有哪些？

1. 模板方法模式（核心适配） #

该模式完全匹配你的场景特性：固定执行步骤、单步骤可独立扩展、整体执行逻辑统一。你当前的单次API调用有请求→提取→保存3个固定执行顺序，刚好符合模板方法的核心设计逻辑：把不变的行为（执行步骤框架、执行顺序）抽离封装到基类，可变的个性化逻辑（不同API的请求参数、提取规则、存储目标）留给子类实现。

参考代码结构示例：

from abc import ABC, abstractmethod

# 抽象基类：定义固定执行模板
class BaseAPIScraper(ABC):
    # 固定执行流程，禁止子类修改
    def run(self):
        resp = self.request()
        target_data = self.extract(resp)
        self.save(target_data)
    
    @abstractmethod
    def request(self):
        """不同API实现自定义请求逻辑（header、参数、签名等）"""
        pass
    
    @abstractmethod
    def extract(self, resp):
        """不同API实现自定义字段提取规则"""
        pass
    
    @abstractmethod
    def save(self, data):
        """不同API实现自定义存储逻辑，通用存储可直接做默认实现"""
        pass

# 每个API单独实现子类，不需要关心执行顺序，只需要补全单步逻辑
class Api1Scraper(BaseAPIScraper):
    def request(self):
        # API1专属请求逻辑
        pass
    def extract(self, resp):
        # API1专属字段提取逻辑
        pass
    def save(self, data):
        # API1专属存储逻辑
        pass

后续新增API只需要新增子类即可，不需要修改原有执行逻辑，符合开闭原则。

2. 搭配策略模式处理异步调度 #

因为你的3个API调用无顺序要求、异步并行执行，可以把每个API的scraper实例作为独立的执行策略，调度层统一收集所有策略后批量异步触发即可：

import asyncio
# 收集所有需要执行的API策略实例
scraper_strategies = [Api1Scraper(), Api2Scraper(), Api3Scraper()]

async def batch_run():
    # 异步并行执行所有策略的run方法
    tasks = [asyncio.to_thread(s.run) for s in scraper_strategies]
    await asyncio.gather(*tasks)

# 定时调度部分直接每5分钟触发batch_run方法即可

额外优化建议 #

如果多个API的提取、存储逻辑有通用部分，可以直接在基类实现通用默认逻辑，子类只有特殊需求时再重写对应方法即可，减少重复代码。比如所有API都存储到同一MySQL实例的不同表，那基类的save方法可以封装通用的数据库连接、写入逻辑，子类只需要返回对应表名和格式化后的数据即可。

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