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

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

目录

1. 为什么多节点协作会出冲突
2. 冲突的典型场景
3. 分布式锁的实现
4. 基于版本的乐观并发控制
5. 任务队列的分区与亲和性
6. 生产环境的落地经验
7. 小结

1. 为什么多节点协作会出冲突

做多节点发送系统的同学大概率都经历过这个阶段：一开始只有一个进程在跑，逻辑简单清晰。后来为了提高吞吐量，启动了多个工作进程（或者部署到了多台机器），结果开始出现奇怪的问题：同一个联系人收到了两条一样的消息，或者两个进程同时抢到了同一条待发送任务。

问题的根源是单机时代的"内存状态即真理"假设在多节点环境下不再成立。每个节点只看到自己的局部状态，看不到其他节点正在干什么。

我们在实际系统里总结过四类由多节点协作不当引发的冲突：

这四类问题指向一个核心需求：需要一种机制让多个节点之间协调好谁做什么、做到哪了。

2. 冲突的典型场景

在深入方案之前，先搞清楚冲突到底发生在哪些环节：

┌─────────────┐ 任务入队 ────▶ │ 共享队列 │ ◀─── 多节点共享的数据源 └──────┬──────┘ │ ┌────────────┼────────────┐ ▼ ▼ ▼ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ Node A │ │ Node B │ │ Node C │ ← 多个工作节点 │ 抢任务？ │ │ 抢任务？ │ │ 抢任务？ │ └────┬────┘ └────┬────┘ └────┬────┘ │ │ │ ▼ ▼ ▼ 发送消息 发送消息 发送消息 │ │ │ └────────────┼────────────┘ ▼ 写回结果到共享存储 │ ⚠️ 这里就是冲突高发区

冲突最密集的三个时刻：

1. 取任务时：多个节点同时查SELECT * FROM queue WHERE status='pending' LIMIT N
2. 写结果时：多个节点同时更新同一条记录的状态
3. 资源访问时：多个节点同时向同一个账号发消息

下面针对这三个时刻分别给解决方案。

3. 分布式锁的实现

核心思路：对关键操作加排他锁，确保同一时间只有一个节点能执行。

SQLite 本身支持文件级锁，但默认的BEGIN IMMEDIATE是数据库级别的（整个库被锁住），粒度太粗。我们需要的是行级或任务级的细粒度锁。

importsqlite3importtimeimportthreadingimportuuidclassDistributedLock:"""基于 SQLite 的轻量分布式锁 适用于 SQLite 单文件、多进程/多线程场景。 如果用 MySQL/PostgreSQL，可以用 SELECT ... FOR UPDATE 或 GET_LOCK()。 """def__init__(self,db_path:str):self.db_path=db_path self._init_table()def_init_table(self):conn=sqlite3.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:str=None,ttl_seconds:int=30)->bool:""" 尝试获取锁。 ttl_seconds: 锁的超时时间（秒），防止持锁方崩溃导致死锁。 返回 True 表示获取成功。 """ifowner_idisNone:owner_id=f"{uuid.uuid4().hex[:8]}"now=time.time()expires=now+ttl_seconds conn=sqlite3.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:"""释放锁（只能释放自己持有的）"""conn=sqlite3.connect(self.db_path)cursor=conn.execute(""" DELETE FROM distributed_locks WHERE lock_key = ? AND owner_id = ? """,(lock_key,owner_id))conn.commit()deleted=cursor.rowcount conn.close()returndeleted>0defis_locked(self,lock_key:str)->bool:"""检查某个锁是否被持有"""conn=sqlite3.connect(self.db_path)row=conn.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_path=db_path self.node_id=node_id self.lock=DistributedLock(db_path)deffetch_with_lock(self,task_id:int)->Optional[dict]:""" 安全地取一条任务：先加锁再查询+更新。 返回任务数据，或 None（已被其他节点取走）。 """lock_key=f"task_{task_id}"ifnotself.lock.acquire(lock_key,self.node_id,ttl_seconds=60):returnNone# 别人已经拿到了try:conn=sqlite3.connect(self.db_path)conn.execute("BEGIN IMMEDIATE")# 双重检查：拿到锁之后再确认任务还是 pending 状态row=conn.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_key=f"task_{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_path=db_pathdeffetch_for_update(self,task_id:int)->Optional[dict]:"""取出任务及当前版本号"""conn=sqlite3.connect(self.db_path)row=conn.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 表示已被其他人修改（需重试）。 """conn=sqlite3.connect(self.db_path)cursor=conn.execute(""" UPDATE queue_messages SET status = ?, _version = _version + 1, updated_at = datetime('now','localtime') WHERE id = ? AND _version = ? """,(new_status,task_id,expected_version))affected=cursor.rowcount conn.commit()conn.close()returnaffected>0

使用方式：节点 A 先读到 version=5 的任务，处理后尝试 UPDATE WHERE version=5。如果此时节点 B 已经把它改成了 version=6，那么 A 的 UPDATE 影响行数为 0，A 知道冲突发生了，可以选择重读最新状态后重试。

乐观 vs 悲观怎么选？

对于我们的消息发送场景，冲突率通常不高（每条消息只被处理一次的概率很高），所以两种策略都可以。我们推荐混合使用：任务领取用分布式锁保证不重复，状态更新用乐观并发减少锁持有时间。

5. 任务队列的分区与亲和性

除了锁和并发控制，还有一个更根本的解决思路：从源头避免冲突，让每个节点只处理自己的那部分数据。

这就是分区 + 亲和性的设计：

classPartitionedRouter:"""基于分区的任务路由器：让不同节点处理不同的任务子集"""def__init__(self,node_id:str,total_nodes:int):self.node_id=node_id self.total_nodes=total_nodesdefowns_task(self,task_id:int)->bool:"""判断本节点是否应该处理该任务"""returntask_id%self.total_nodes==self._node_index@propertydef_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:int=50)->list:"""只拉取属于自己的任务"""conn=sqlite3.connect(db_path)# 方式一：按 ID 取模筛选（最简单，但可能不均匀）rows=conn.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_id=node_id self._affinity_cache={}# conversation_id → node_iddefassign_node(self,conversation_id:str,available_nodes:list)->str:""" 为一个会话分配负责节点。 同一会话始终分配到同一个节点，除非那个节点挂了。 """ifconversation_idinself._affinity_cache:assigned=self._affinity_cache[conversation_id]ifassignedinavailable_nodes:returnassignedelse:# 原分配的节点不在了，重新分配delself._affinity_cache[conversation_id]# 简单轮询分配（也可以用一致性哈希）idx=hash(conversation_id)%len(available_nodes)chosen=available_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 文件锁 + 版本号对中小规模足够用了。