共计 1450 个字符，预计需要花费 4 分钟才能阅读完成。

前言

最近在 Allegro 平台上开发自动化任务时，发现现有脚本经常因为 API 限流、网络波动等问题中断，手动恢复又特别耗时。经过几轮迭代，终于总结出一套高可靠的 Skill 程序开发方案，这里把核心思路和关键代码分享给大家。

一、为什么需要重构自动化任务

1. 传统轮询模式的三大痛点

• API 调用频率限制（Rate Limiting）：Allegro 的 REST API 对每分钟请求数有严格限制，传统轮询容易触发 429 错误
• 任务状态持久化 ：脚本崩溃后无法从断点恢复，需要重新执行整个流程
• 错误恢复机制缺失 ：网络抖动或 API 异常时缺乏重试策略，导致任务直接失败

2. 事件驱动架构的优势

对比传统每 5 分钟轮询的方案，基于事件驱动（Event-Driven）的设计可以：

1. 通过 Webhook 接收库存变更等实时事件
2. 使用消息队列缓冲突发请求
3. 减少无效 API 调用达 70% 以上

二、核心架构实现

1. 整体流程图

flowchart TD
    A[OAuth2.0 认证] --> B[请求限流器]
    B --> C{API 调用}
    C -->| 成功 | D[处理响应]
    C -->| 失败 | E[错误分类处理]
    E -->| 可重试 | B
    E -->| 致命错误 | F[持久化状态]

2. 关键模块代码

OAuth2.0 认证封装

class AllegroAuth:
    """自动处理 token 刷新的认证模块"""
    def __init__(self, client_id: str, client_secret: str):
        self._token: Optional[dict] = None
        self._expires_at: float = 0

    async def get_token(self) -> str:
        if time.time() > self._expires_at - 60:  # 提前 1 分钟刷新
            await self._refresh_token()
        return self._token["access_token"]

令牌桶限流器

class TokenBucket:
    """基于 asyncio 的令牌桶算法实现"""
    def __init__(self, rate: int, capacity: int):
        self._tokens = capacity
        self._last_check = time.monotonic()

    async def acquire(self):
        while self._tokens < 1:
            await asyncio.sleep(0.1)
            self._update_tokens()
        self._tokens -= 1

三、避坑经验

1. 特定错误码处理

• 5031 错误（库存冲突）：需要实现乐观锁机制
• 429 错误（限流）：动态调整令牌桶容量

2. 分布式锁实现

# 使用 Redis 实现跨进程锁
async with redis.lock("order_123", timeout=10):
    await process_order()

四、性能验证

使用 Locust 进行压力测试：

1. 优化前：QPS 15，平均延迟 2 秒
2. 优化后：QPS 45，平均延迟 800ms

五、扩展思考

这套架构稍作改造就能支持 eBay、Amazon 等平台：

1. 抽象出平台适配层（Platform Adapter）
2. 使用策略模式切换不同 API 规范
3. 尝试用 AWS Lambda 实现无服务化改造

结语

实际部署后发现这套方案最实用的三个特性：

• 令牌桶算法让 API 调用量始终保持在安全线内
• 状态持久化功能让凌晨的定时任务不再需要人工干预
• 详细的 Prometheus 指标方便定位性能瓶颈

完整代码已上传 GitHub（示例仓库：allegro-skills-demo），欢迎交流优化建议！