WhatsApp 多节点协同调度中的分布式冲突避免

WhatsApp 多节点协同调度中的分布式冲突避免目录为什么多节点协作会出冲突冲突的典型场景分布式锁的实现基于版本的乐观并发控制任务队列的分区与亲和性生产环境的落地经验小结1. 为什么多节点协作会出冲突做多节点发送系统的同学大概率都经历过这个阶段一开始只有一个进程在跑逻辑简单清晰。后来为了提高吞吐量启动了多个工作进程或者部署到了多台机器结果开始出现奇怪的问题同一个联系人收到了两条一样的消息或者两个进程同时抢到了同一条待发送任务。问题的根源是单机时代的内存状态即真理假设在多节点环境下不再成立。每个节点只看到自己的局部状态看不到其他节点正在干什么。我们在实际系统里总结过四类由多节点协作不当引发的冲突冲突类型表现后果重复发送两个节点同时读取到同一条 pending 消息各自发了一遍客户收到重复消息体验差任务丢失节点 A 取走任务后崩溃任务没有回到队列消息永远发不出去计数不一致各节点独立统计发送量汇总后总数对不上运营数据不可信资源竞争多个节点同时操作同一账号触发平台限速账号被限制这四类问题指向一个核心需求需要一种机制让多个节点之间协调好谁做什么、做到哪了。2. 冲突的典型场景在深入方案之前先搞清楚冲突到底发生在哪些环节┌─────────────┐ 任务入队 ────▶ │ 共享队列 │ ◀─── 多节点共享的数据源 └──────┬──────┘ │ ┌────────────┼────────────┐ ▼ ▼ ▼ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ Node A │ │ Node B │ │ Node C │ ← 多个工作节点 │ 抢任务 │ │ 抢任务 │ │ 抢任务 │ └────┬────┘ └────┬────┘ └────┬────┘ │ │ │ ▼ ▼ ▼ 发送消息 发送消息 发送消息 │ │ │ └────────────┼────────────┘ ▼ 写回结果到共享存储 │ ⚠️ 这里就是冲突高发区冲突最密集的三个时刻取任务时多个节点同时查SELECT * FROM queue WHERE statuspending LIMIT N写结果时多个节点同时更新同一条记录的状态资源访问时多个节点同时向同一个账号发消息下面针对这三个时刻分别给解决方案。3. 分布式锁的实现核心思路对关键操作加排他锁确保同一时间只有一个节点能执行。SQLite 本身支持文件级锁但默认的BEGIN IMMEDIATE是数据库级别的整个库被锁住粒度太粗。我们需要的是行级或任务级的细粒度锁。importsqlite3importtimeimportthreadingimportuuidclassDistributedLock:基于 SQLite 的轻量分布式锁 适用于 SQLite 单文件、多进程/多线程场景。 如果用 MySQL/PostgreSQL可以用 SELECT ... FOR UPDATE 或 GET_LOCK()。 def__init__(self,db_path:str):self.db_pathdb_path self._init_table()def_init_table(self):connsqlite3.connect(self.db_path)conn.execute( CREATE TABLE IF NOT EXISTS distributed_locks ( lock_key TEXT PRIMARY KEY, -- 锁的名称如 task_123 owner_id TEXT NOT NULL, -- 持有者标识 acquired_at TEXT NOT NULL, expires_at TEXT NOT NULL, -- 过期时间防死锁 meta TEXT DEFAULT ) )conn.commit()conn.close()defacquire(self,lock_key:str,owner_id:strNone,ttl_seconds:int30)-bool: 尝试获取锁。 ttl_seconds: 锁的超时时间秒防止持锁方崩溃导致死锁。 返回 True 表示获取成功。 ifowner_idisNone:owner_idf{uuid.uuid4().hex[:8]}nowtime.time()expiresnowttl_seconds connsqlite3.connect(self.db_path)# 先清理过期的锁兜底清理conn.execute( DELETE FROM distributed_locks WHERE expires_at datetime(now,localtime) )# 尝试插入新锁如果 lock_key 已存在则失败try:conn.execute( INSERT INTO distributed_locks (lock_key, owner_id, acquired_at, expires_at) VALUES (?, ?, datetime(now,localtime), datetime(now,localtime, ? || seconds)) ,(lock_key,owner_id,str(ttl_seconds)))conn.commit()conn.close()returnTrueexceptsqlite3.IntegrityError:# lock_key 已存在说明别人持有conn.close()returnFalsedefrelease(self,lock_key:str,owner_id:str)-bool:释放锁只能释放自己持有的connsqlite3.connect(self.db_path)cursorconn.execute( DELETE FROM distributed_locks WHERE lock_key ? AND owner_id ? ,(lock_key,owner_id))conn.commit()deletedcursor.rowcount conn.close()returndeleted0defis_locked(self,lock_key:str)-bool:检查某个锁是否被持有connsqlite3.connect(self.db_path)rowconn.execute(SELECT 1 FROM distributed_locks WHERE lock_key ?,(lock_key,)).fetchone()conn.close()returnrowisnotNoneclassLockingTaskFetcher:带锁的任务取用器def__init__(self,db_path:str,node_id:str):self.db_pathdb_path self.node_idnode_id self.lockDistributedLock(db_path)deffetch_with_lock(self,task_id:int)-Optional[dict]: 安全地取一条任务先加锁再查询更新。 返回任务数据或 None已被其他节点取走。 lock_keyftask_{task_id}ifnotself.lock.acquire(lock_key,self.node_id,ttl_seconds60):returnNone# 别人已经拿到了try:connsqlite3.connect(self.db_path)conn.execute(BEGIN IMMEDIATE)# 双重检查拿到锁之后再确认任务还是 pending 状态rowconn.execute( SELECT * FROM queue_messages WHERE id ? AND status pending ,(task_id,)).fetchone()ifnotrow:conn.rollback()conn.close()self.lock.release(lock_key,self.node_id)returnNone# 标记为处理中conn.execute( UPDATE queue_messages SET status sending, updated_at datetime(now,localtime) WHERE id ? ,(task_id,))conn.commit()conn.close()returndict(row)finally:# 注意这里不释放锁锁应该在任务完成后再释放# 防止其他节点在我们处理期间抢同一任务passdefcomplete_task(self,task_id:int,success:bool):完成任务后释放锁lock_keyftask_{task_id}self.lock.release(lock_key,self.node_id)# 同时更新任务状态...坑点TTL超时时间必须设且不能太长。这是防死锁的关键。如果持锁节点在处理过程中崩了锁永远不会被主动释放。TTL 到期后下次任何人 acquire 时会先清理过期锁这样死掉的节点持有的锁会被自动回收。建议 TTL 设为正常处理耗时的 2–3 倍比如处理通常 10 秒完成TTL 设 30 秒。BEGIN IMMEDIATE很重要。SQLite 默认是 deferred transaction第一次读不会加写锁。用IMMEDIATE模式会在 BEGIN 时就获取 RESERVED 锁防止两个事务同时在读-改-写的路径上产生竞态。4. 基于版本的乐观并发控制分布式锁是悲观策略先锁再做另一种思路是乐观策略不加锁但在提交时检测是否有人改过了。classOptimisticTaskUpdater:基于版本号的乐观并发更新器def__init__(self,db_path:str):self.db_pathdb_pathdeffetch_for_update(self,task_id:int)-Optional[dict]:取出任务及当前版本号connsqlite3.connect(self.db_path)rowconn.execute( SELECT *, _version FROM queue_messages WHERE id ? ,(task_id,)).fetchone()conn.close()returndict(row)ifrowelseNonedefupdate_with_version(self,task_id:int,expected_version:int,new_status:str,**kwargs)-bool: 带版本校验的更新。 只有当当前版本 expected_version 时才成功。 返回 True 表示更新成功False 表示已被其他人修改需重试。 connsqlite3.connect(self.db_path)cursorconn.execute( UPDATE queue_messages SET status ?, _version _version 1, updated_at datetime(now,localtime) WHERE id ? AND _version ? ,(new_status,task_id,expected_version))affectedcursor.rowcount conn.commit()conn.close()returnaffected0使用方式节点 A 先读到 version5 的任务处理后尝试 UPDATE WHERE version5。如果此时节点 B 已经把它改成了 version6那么 A 的 UPDATE 影响行数为 0A 知道冲突发生了可以选择重读最新状态后重试。乐观 vs 悲观怎么选场景推荐策略原因冲突频率低5%乐观并发不加锁开销小偶尔重试可接受冲突频率高20%分布式锁频繁重试反而浪费更多资源关键路径不允许重复执行分布式锁重复执行的代价太高如扣款对于我们的消息发送场景冲突率通常不高每条消息只被处理一次的概率很高所以两种策略都可以。我们推荐混合使用任务领取用分布式锁保证不重复状态更新用乐观并发减少锁持有时间。5. 任务队列的分区与亲和性除了锁和并发控制还有一个更根本的解决思路从源头避免冲突让每个节点只处理自己的那部分数据。这就是分区 亲和性的设计classPartitionedRouter:基于分区的任务路由器让不同节点处理不同的任务子集def__init__(self,node_id:str,total_nodes:int):self.node_idnode_id self.total_nodestotal_nodesdefowns_task(self,task_id:int)-bool:判断本节点是否应该处理该任务returntask_id%self.total_nodesself._node_indexpropertydef_node_index(self)-int:# 把 node_id 映射到 0 ~ total_nodes-1 的整数# 简单实现取 hash 最后几位 modreturnhash(self.node_id)%self.total_nodesdefget_own_tasks(self,db_path:str,limit:int50)-list:只拉取属于自己的任务connsqlite3.connect(db_path)# 方式一按 ID 取模筛选最简单但可能不均匀rowsconn.execute( SELECT * FROM queue_messages WHERE status pending AND id % ? ? ORDER BY id ASC LIMIT ? ,(self.total_nodes,self._node_index,limit)).fetchall()conn.close()return[dict(r)forrinrows]classAffinityScheduler:带亲和性的调度器同一会话的消息尽量由同一节点处理def__init__(self,node_id:str):self.node_idnode_id self._affinity_cache{}# conversation_id → node_iddefassign_node(self,conversation_id:str,available_nodes:list)-str: 为一个会话分配负责节点。 同一会话始终分配到同一个节点除非那个节点挂了。 ifconversation_idinself._affinity_cache:assignedself._affinity_cache[conversation_id]ifassignedinavailable_nodes:returnassignedelse:# 原分配的节点不在了重新分配delself._affinity_cache[conversation_id]# 简单轮询分配也可以用一致性哈希idxhash(conversation_id)%len(available_nodes)chosenavailable_nodes[idx]self._affinity_cache[conversation_id]chosenreturnchosen分区的好处是显而易见的如果节点 A 只处理 ID % 3 0 的任务节点 B 只处理 ID % 3 1 的任务它们之间天然不会冲突。这比任何锁机制的效率都高因为根本不需要协调。亲和性的好处在于上下文连续性同一个对话的所有消息都由同一个节点处理这个节点的内存里可以缓存该对话的上下文信息比如用户上一句说了什么不需要每次都从数据库加载。当然分区的代价是负载可能不均匀。如果某一批任务的 ID 刚好都落在同一个分区里对应节点就会忙死而其他节点闲着。解决方法是用一致性哈希代替简单取模或者在取模之前先 shuffle 任务 ID。6. 生产环境的落地经验分布式冲突避免的理论方案很多但实际落地时怎么取舍很重要。我们以 WAWarmer 的多节点调度模块为例看它是怎么做冲突管理的。它采用的是分层防御的策略第一层是任务分区按 conversation_id 做一致性哈希让不同会话归属不同节点大部分情况下各节点各管各的天然无冲突第二层是在跨节点操作比如全局统计、配置热更新时用 SQLite 的 IMMEDIATE 事务做悲观锁保护第三层是所有状态更新带版本号做乐观校验即使前两层漏掉了也能在提交时发现冲突。这种分层的好处是没有单一依赖。分区失效了比如某个节点临时下线要 rehash还有锁兜底锁性能不够好了还有乐观并发保底。三层同时出问题的概率极低。另外它在监控层面做了一个简单的冲突计数器每次乐观更新失败version 不匹配或锁获取失败时计数1超过阈值就告警。这个指标上线后帮我们提前发现了两次因网络抖动导致的异常高冲突率在真正影响业务之前就做了干预。7. 小结多节点协同这件事核心就是让多个工作者别互相踩脚。三件事隔离优先能用分区就用分区从源头让各节点操作不同数据集锁保底线跨节点临界操作必须加锁且锁必须有 TTL 防死锁版本兜底状态更新带上版本号即使锁不够细也能发现冲突。如果你的系统还在单进程跑但已经开始有扩展计划建议按这三步演进先把任务模型加上_version字段和owner_node字段成本很低但后续有用再上任务分区路由一致性哈希或取模让多节点并行时不冲突最后补分布式锁和冲突监控做到可观测、可回溯。整套从单节点到多节点的改造大约 2–3 个工作日其中调试分区均衡性和锁超时参数会占主要时间。起步阶段没必要上 ZooKeeper 或 etcd 这种重量级协调服务SQLite 文件锁 版本号对中小规模足够用了。