基于 DynamoDB 条件表达式构建可靠的分布式任务代理

最近我在整理手头的一套分布式任务系统时把基于 DynamoDB 的组件都攒成了一个系列来维护其中 ThunderAgent 这个代理模块是这次重构里最顺手的一部分。它本身不复杂但解决了我之前用传统关系型数据库做任务队列时遇到的锁竞争、重试困难、状态混乱一堆问题。如果你也在做后台任务调度、事件驱动处理或者只是想找个轻量方案把异步任务管起来这篇文章应该能给你一些可以直接抄作业的思路。ThunderAgent 本质上是挂在 DynamoDB 上的一个“跑腿”程序它负责从表里捞待处理的任务执行对应业务逻辑再把结果状态写回去。整个过程围绕 DynamoDB 的原子操作和条件表达式展开所以天然适合做分布式环境下的任务分发。下面我会从设计思路、表结构到实际代码一步步拆开讲顺便把我在生产环境里踩过的坑也都列出来。1. 项目概述ThunderAgent 在 Dynamo 系列中的定位Dynamo 系列是我这边一组围绕 DynamoDB 的工具集合里面包括数据迁移器、状态机辅助模块还有今天要说的 ThunderAgent。每个组件都尽量保持单一职责只围绕 DynamoDB 的某几个特性做深做透。ThunderAgent 的职责非常清晰消费任务、执行业务、更新状态整个生命周期都通过一张任务表来驱动。说“代理”而不是“队列”是因为它不像 SQS 那样只负责消息传递而是真正参与任务执行的过程。你可以在同一个代码里定义执行逻辑也可以把逻辑放到后台线程里ThunderAgent 只保证“拿得出来、推得回去”。对于需要重试、限流、断点续跑的场景这种方式比单纯把消息丢进队列要可控得多。当初做这个模块的背景很直接——我需要一个自动处理报表生成的工具任务多的时候每分钟能涌进来上千个少的时候可能一整天都没有一个。用固定线程池吗会浪费资源。用消息队列吗又要额外维护一套 infra。后来发现 DynamoDB 的按需扩容和条件写入正好能满足这种突发式需求于是干脆把任务状态全交给表来管理ThunderAgent 在里面只当一个勤恳的搬运工。1.1 核心需求解析围绕 ThunderAgent我梳理了四个核心需求任务生命周期管理、分布式环境下的并发控制、失败重试机制、以及执行结果的可追踪性。这四点看起来基础但真的全做好并不容易。任务生命周期我定义了五个状态PENDING待处理、RUNNING处理中、SUCCESS成功、FAILED失败、DONE终态。每次任务进来先落库为 PENDING代理获取后置为 RUNNING执行完毕根据结果更新为 SUCCESS 或 DONE。如果执行异常会根据重试策略回到 PENDING 或者直接置为 FAILED。并发控制是这里面最需要小心的部分。多个代理实例同时启动时很容易出现同一任务被两个实例拿到的竞态问题。DynamoDB 的条件表达式在这里起到了“锁”的作用——更新状态时加上status :pending的条件只有满足这个条件才能成功写入否则表示任务已经被别的实例拿走了。这相当于数据库层面的乐观锁足够应对大多数场景。1.2 为什么选择 DynamoDB 作为底座选型那会儿我也犹豫过要不要直接用 Redis 或者 MySQL 来处理任务状态。但综合对比下来DynamoDB 有几个没法拒绝的优势。第一是无服务器架构不需要操心实例维护按读取写入量计费对于任务量忽高忽低的情况非常友好。第二是条件表达式和原子计数器这两者能帮我实现并发去重和状态流转不用自己在代码里造锁。第三是有 TTL 功能可以自动清理几周前的历史任务省心不少。当然 DynamoDB 也有它的短板比如读写吞吐量不够时会有节流报错但你可以在代理端做退避重试来缓解。还有它不擅长复杂的聚合查询但这正好不是我要用它的地方。我的原则是任务状态走 DynamoDB业务数据继续放关系型库各管各的。2. 技术选型与设计思路DynamoDB 的 API 本身很简单难的是怎么用对姿势。在设计 ThunderAgent 的架构时我重点考虑了两个方案事件驱动和主动轮询。事件驱动是指通过 DynamoDB Streams 来触发 Lambda 或代理实例这种方式的实时性很好但最大的问题是如果业务执行时间较长则容易拖垮 Lambda 的并发限制而且需要额外管理流数据的消费位置。主动轮询则是让代理每隔几秒查一次表找到满足条件的状态再处理。虽然实时性略差但胜在实现简单、流程可控也方便处理重试和超时。最终我选择了主动轮询原因只有一个我要在代理端做灵活的重试逻辑而轮询模式下执行逻辑和代理代码是同步的调试起来非常顺手。实测下来每秒处理几百个任务是绰绰有余的绝大多数业务场景完全够用。2.1 表结构设计细节我建了一张名为TaskTable的任务表主键是taskIdStringUUIDsort key 没有使用因为任务之间没有排序需求。需要说明的是为了支持查询 PENDING 状态的任务我额外加了一个全局二级索引GSI索引的主键用了statussort key 用了createdAt。这样代理每次只需要query这个 GSI就能快速拿到一批最早的待处理任务。下面是我实际用的表结构定义字段不算多但每个都有用途字段名类型说明taskIdString主键任务唯一标识statusStringPENDING / RUNNING / SUCCESS / DONE / FAILEDpayloadMap任务携带的业务数据retryCountNumber当前已重试次数maxRetryNumber最大允许重试次数lockTimeNumber租约过期时间戳用于任务超时释放createdAtNumber任务创建时间epoch 秒updatedAtNumber最近一次状态更新时间关于lockTime这个字段我用它来防止任务在执行过程中代理突然崩溃导致的状态卡死。代理将任务置为 RUNNING 时同时写入一个lockTime now 300s如果超过这个时间状态还没变就可以认为任务已经“死掉”允许其他代理重新接管。2.2 代理端的轮询策略轮询不是简单死循环我做了几个优化。首先通过 GSI 查询时每次最多返回 10 个 PENDING 任务避免一批拉太多导致处理不过来。其次每次查询之后根据任务量情况动态调整休眠时间如果上一批有积压就缩短到 1 秒反之则延长到 5 秒。这个简单启发式策略能让代理在高负载下迅速响应在空闲时又不至于浪费查询成本。还有一个细节是在多代理实例同时运行时单靠status :pending条件更新并不能完全避免重复执行。因为两个实例可能同时查询到同一个 PENDING 任务然后在各自进程内并发更新其中只会有一个成功。为此我同时校验了lockTime让它小于当前时间这样即使同一个任务在当前没有超时其他代理也无法抢占。代码如下面的示例稍后会详细解读。3. 核心实操从零搭建 ThunderAgent实操部分我用自己的环境为例系统是 Ubuntu 22.04Python 3.10依赖boto3和schedule库。DynamoDB 我直接用了 AWS 云上的表如果你只是想本地测试可以用dynamodb-localjar 包跑一个模拟环境连接方式稍微改一下即可。创建任务表的命令并不复杂需要注意的地方是预置容量和索引配置。我建表时用了按需计费模式不用预置读写容量省去了一开始估容量的问题。索引则必须提前设置好因为 DynamoDB 不支持事后直接加 GSI 的某些属性覆盖需要重新建表或者迁移。3.1 环境准备首先安装 Python SDKpip install boto3接着创建任务表用 boto3 的 Python 脚本大概是这样的import boto3 dynamodb boto3.resource(dynamodb, region_nameus-east-1) table dynamodb.create_table( TableNameTaskTable, KeySchema[ {AttributeName: taskId, KeyType: HASH} ], AttributeDefinitions[ {AttributeName: taskId, AttributeType: S}, {AttributeName: status, AttributeType: S}, {AttributeName: createdAt, AttributeType: N} ], GlobalSecondaryIndexes[ { IndexName: status-createdAt-index, KeySchema: [ {AttributeName: status, KeyType: HASH}, {AttributeName: createdAt, KeyType: RANGE} ], Projection: {ProjectionType: ALL} } ], BillingModePAY_PER_REQUEST ) table.wait_until_exists() print(Table created successfully)这里的关键点是把 GSI 的查询键设计成status createdAt因为总想按状态取最早的任务。ProjectionType用的ALL这样代理从索引里就能拿到全部字段不用再去查主表减少一次网络往返。3.2 核心代理代码实现ThunderAgent 的核心代码其实只有两个函数fetch_pending_tasks和process_task。前者负责从索引获取待处理任务后者则负责执行业务逻辑并将结果状态回写。这里我把最关键的循环和条件更新都列出来。import time import uuid import boto3 from boto3.dynamodb.conditions import Attr dynamodb boto3.resource(dynamodb, region_nameus-east-1) table dynamodb.Table(TaskTable) def fetch_pending_tasks(batch_size10): response table.query( IndexNamestatus-createdAt-index, KeyConditionExpression#s :status, ExpressionAttributeNames{#s: status}, ExpressionAttributeValues{:status: PENDING}, Limitbatch_size ) return response.get(Items, []) def execute_business_logic(task): # 这里是实际业务比如生成报表、调用接口等 # 这里只模拟一个耗时操作 time.sleep(2) return {result: ok, task_id: task[taskId]} def process_task(task): # 1. 尝试通过条件更新将任务状态从 PENDING 改为 RUNNING # 同时设置 lockTime 为当前时间 300 秒 now int(time.time()) try: update_response table.update_item( Key{taskId: task[taskId]}, UpdateExpressionSET #s :running, lockTime :lock, updatedAt :updated, ConditionExpressionAttr(#s).eq(PENDING) Attr(lockTime).lt(now), ExpressionAttributeNames{#s: status}, ExpressionAttributeValues{ :running: RUNNING, :lock: now 300, :updated: now }, ReturnValuesALL_NEW ) except ClientError as e: if e.response[Error][Code] ConditionalCheckFailedException: print(f任务 {task[taskId]} 已被其他实例抢占跳过) return raise # 2. 执行真正的业务逻辑 try: result execute_business_logic(task) # 业务成功更新状态为 SUCCESS 并清理 lockTime table.update_item( Key{taskId: task[taskId]}, UpdateExpressionSET #s :success, updatedAt :updated REMOVE lockTime, ExpressionAttributeNames{#s: status}, ExpressionAttributeValues{ :success: SUCCESS, :updated: int(time.time()) } ) except Exception as e: # 业务失败重试数量 1如果没超过最大重试就回到 PENDING retry_count task.get(retryCount, 0) 1 if retry_count task.get(maxRetry, 3): table.update_item( Key{taskId: task[taskId]}, UpdateExpressionSET #s :pending, retryCount :rc, updatedAt :updated REMOVE lockTime, ExpressionAttributeNames{#s: status}, ExpressionAttributeValues{ :pending: PENDING, :rc: retry_count, :updated: int(time.time()) } ) else: table.update_item( Key{taskId: task[taskId]}, UpdateExpressionSET #s :failed, updatedAt :updated REMOVE lockTime, ExpressionAttributeNames{#s: status}, ExpressionAttributeValues{ :failed: FAILED, :updated: int(time.time()) } ) def main_loop(): while True: pending_tasks fetch_pending_tasks(batch_size10) for task in pending_tasks: process_task(task) # 动态休眠 time.sleep(1 if len(pending_tasks) 0 else 5) if __name__ __main__: main_loop()这个代码里的ConditionExpression是 ThunderAgent 的灵魂。我用Attr(#s).eq(PENDING)保证只有状态等于 PENDING 的任务才能被拉走用Attr(lockTime).lt(now)保证只有锁过期或者根本没有锁的任务才能被接管。这两条合起来即使有十个代理实例同时运行也不会抢同一份工作。3.3 参数设置与调优建议我在实际部署中把batch_size从默认的 10 调高到了 30因为我们的单个任务通常只需几秒30 个的并发粒度在这个水平上处理得刚刚好。但是要注意Limit不是一次扫描的上限而是返回记录数的上限如果表里 PENDING 任务特别多一次只拿 30 条也能保证响应速度。休眠时间是一个值得调优的参数。我原来的机器上用的是固定 3 秒后来发现低峰期白白浪费查询费用。改成动态之后高峰期处理速度基本没感觉低峰期成本降了不少。如果你的任务时延要求很高建议把空闲休眠时间降到 1 秒代价是查询频率会拉高需要自己权衡。另外关于任务超时时间lockTime我一开始设的是 60 秒结果线上有个任务处理超过 2 分钟被其他代理提前接管导致重复执行。后来我改成 300 秒并且把正常执行时间放在日志里做监控一旦超时就能及时调整上限。这个值最好是你业务耗时的 1.5 倍以上留足缓冲。4. 常见问题与排查技巧实录不管代码写得再稳生产环境总会给你来点意外。我在跑 ThunderAgent 的过程中遇到过不少问题这里挑几个典型的连同排查思路和解决方案一起放在表格里方便你以后对照着查。4.1 高频问题速查表症状可能原因解决办法任务一直卡在 PENDING代理也不取走GSI 状态索引没建好或者容量不够检查索引status-createdAt-index是否存在用表格里的查询语句手动验证同一个任务被多个代理执行lockTime 未正确更新或条件表达式漏了 lockTime 判断确认条件表达式里有lockTime now并检查更新后是否返回了ALL_NEW代理日志里有大量ConsumedCapacity报错DynamoDB 节流尤其是 GSI 容量耗尽将表切换为按需计费模式或者在代理端加入退避重试半夜任务堆积但代理没反应代理进程被杀或代码假死给代理增加心跳日志并部署进程守护工具如 systemd 或 supervisor任务执行很快但状态不更新更新语句条件不满足比如状态不是 RUNNING在执行业务前后打印状态查是否被其他实例改动了4.2 避坑技巧与心得第一必须在代理里加异常兜底。我刚开始只处理了ConditionalCheckFailedException但忽略了网络超时等别的ClientError结果代理进程直接崩掉了。后来我把execute_business_logic整个包进 try/except把所有异常都纳入重试流程这才稳定下来。第二DynamoDB 的 GSI 查询默认是最终一致的。这意味着你在创建任务之后立刻去查询 PENDING有可能因为索引同步延迟而看不到新任务。解决方法是写完之后等待几十毫秒再让代理去轮询或者直接通过主键get_item来校验。如果业务允许可以在创建任务的响应里把 taskId 存到本地缓冲代理优先处理缓冲中的任务。第三轮询模式下要小心日志泛滥。每次都打印拉到的任务列表量大的时候日志会刷得飞起。我把日志级别设定为只在任务状态变更时记录比如成功、失败、重试其他时间仅记录累计计数。这个改动让我排障时清晰不少。第四关于成本控制。DynamoDB 按需计费模式没有预置成本但如果你的表长期高频轮询读写费用也会慢慢起来。我在低峰期会把轮询休眠时间调到 10 秒高峰期自动降回 1 秒这样一个月下来能省不少钱。推荐你像我一样在代理配置里把这个值做成环境变量方便动态调整。4.3 多实例并发与幂等保证如果你打算部署两个或更多的 ThunderAgent 实例一定要测试好并发抢占逻辑。我这边假装开了五个进程同时跑用一百万条任务做了压测最终重复执行率控制在 0.01% 以下。剩余的重试基本来自任务真正常时间执行因为我们的 lockTime 只有 300 秒而有个别报表任务能撑到十分钟以上。针对这类超长任务我最后的解法是把 lockTime 改成每次心跳续期的方式。代理在执行过程中每 60 秒就会调用一次UpdateExpression把lockTime拉高 60 秒这样只有代理真正崩溃超时才会被触发。如果只是任务本身慢它自己会不停续约其他代理永远抢不走。幂等性方面除了状态机本身我还在execute_business_logic里加入了业务去重逻辑比如根据 taskId 判断是否生成过结果文件如果已存在就直接返回成功。双重保险下来就算真的偶发重复执行也不会产生重复数据或重复调用外部接口。5. 后续扩展从单个代理到代理集群ThunderAgent 一开始只是作为单体进程跑的后来任务量上来之后我把它做成了集群模式。核心思想很简单让多个代理实例共享同一张任务表依靠 DynamoDB 的条件表达式自然分散负载。实现上没有任何特殊框架只需把它部署在多台服务器上保持相同的配置即可。在这个模式下我特意做了两个优化。第一是每个代理启动时设置随机的初始化延迟比如 1-3 秒避免所有实例同时哄抢第一波任务。第二是定时输出每个实例的领取任务数方便观察负载是否均衡。实测在四节点集群下分配比例误差不超过 5%表现非常稳定。另外如果任务之间有前后依赖关系比如需要先导入数据再生成报表我会在 payload 里增加一个dependsOn字段代理看到这个字段就暂时不处理而是等待依赖任务的状态变为 SUCCESS。因为 DynamoDB 的 GSI 索引原因等的方式我用的是把任务重置为 PENDING 并设置较远的 lockTime这样其他代理不会拿它当前任务会在下一轮继续检查依赖状态。虽然看起来多了一次查询但逻辑非常清晰。我个人在后面实际使用中有个很深的体会不要把代理想得太复杂它是快递员不是哲学家。快递员只要做好“取件、送件、签收”这几个动作再加上一套可靠的丢失重派机制就已经能扛住绝大多数场景。ThunderAgent 的核心就是把快递员的责任分配清楚让 DynamoDB 替你记住每一单的状态其他的交给时间去跑。