乱序、部分失败与一致性

更新 2026-10-05

我的学习 →

正在读取学习记录…

分布式工作流让多个进程或服务共同完成一项任务。发送方按顺序发出的消息，可能因为网络延迟、失败重发和多条通道而乱序到达；一个工作者失联后，另一个接手，原工作者又可能稍后恢复。系统需要根据任务标识、版本和所有权判断哪些结果仍然有效。

一致性描述不同节点看到的状态能否按约定保持相符。报告任务先运行再完成，页面却先收到完成消息时，不能简单按到达顺序覆盖。下面分别观察消息版本整理和旧工作者晚到的两类问题。

固定发送一条乱序消息和一次晚到写入

报告状态先产生版本1的 running，再产生版本2的 completed，接收端却先拿到版本2。程序把它暂存，收到版本1后连续应用两条消息。

随后给两个工作者依次发放编号1、2。新工作者先保存结果，旧工作者再提交自己的结果。这个随所有权转移而递增的编号叫栅栏令牌，写入端用它拒绝过期执行者。

工作者（worker）就是领取并执行任务的程序进程。例如一台机器负责生成报告，调度程序负责把任务分给它。“失联”表示调度程序暂时收不到它的消息；接手的另一台机器会获得新的执行编号。

先按时间看这两段独立过程：

时刻	消息乱序：发生顺序与收到顺序不同	执行者交接：谁还能保存结果
1	发送版本1：报告正在生成	工作者甲领取任务，拿到令牌1
2	发送版本2：报告已经完成	甲失联，工作者乙接手，拿到令牌2
3	接收端先收到版本2，放入待处理区	乙提交结果，存储端接受令牌2
4	收到版本1，依次处理1和2	甲晚到，存储端发现令牌1已过期

前一段按消息版本恢复顺序，后一段按执行编号决定写入资格。下面的两个类分别处理它们：Projection 整理消息，FencedResult 保存当前执行者的结果。

Projection 在这里表示“从事件整理出来的当前状态”。例如应用收到运行与完成事件后，报告列表中显示的状态就是 completed。

甲持令牌1失联，乙持令牌2接手并保存结果；甲晚到提交时被存储端拒绝 — 时间从上往下读。旧执行者可能继续运行，存储端在保存结果时检查它的执行编号。

整理乱序消息并阻挡旧执行者，顺序、重复意图和执行所有权需要不同字段 — 图1：顺序、重复意图和执行所有权需要不同字段。

保存完整程序并运行

先在 PowerShell 创建并进入学习目录：

mkdir "F:\学习\l4-distributed" -Force
cd "F:\学习\l4-distributed"

保存下面的文件后，在这个终端运行 dir 核对文件名，再执行本节给出的运行命令。

新建 F:\学习\l4-distributed 文件夹，把下面完整代码保存为 distributed_demo.py，然后在PowerShell运行：

cd "F:\学习\l4-distributed"
ls
python -X utf8 distributed_demo.py

先在 PowerShell 创建并进入学习目录：

mkdir "F:\学习\l4-distributed" -Force
cd "F:\学习\l4-distributed"

保存下面的文件后，在这个终端运行 dir 核对文件名，再执行本节给出的运行命令。

"""按版本整理跨节点乱序消息，并拒绝过期工作者的写入。"""


class Projection:
    def __init__(self):
        self.version = 0
        self.status = "new"
        self.pending = {}
        self.seen = {}

    def receive(self, version, status):
        if type(version) is not int or version < 1:
            raise ValueError("版本应为正整数")
        if version in self.seen or version in self.pending:
            saved = self.seen.get(version, self.pending.get(version))
            if saved != status:
                raise ValueError("同版本事件冲突")
            return "duplicate"
        self.pending[version] = status
        # 缺失前序事件时保留后序消息，直到版本连续。
        while self.version + 1 in self.pending:
            next_version = self.version + 1
            self.status = self.pending.pop(next_version)
            self.seen[next_version] = self.status
            self.version = next_version
        return "applied" if version <= self.version else "buffered"


class FencedResult:
    def __init__(self):
        self.token = 0
        self.value = None

    def acquire(self):
        # 每次重新分配所有权时生成更大的栅栏编号。
        self.token += 1
        return self.token

    def write(self, token, value):
        if token != self.token:
            return "stale_worker"
        self.value = value
        return "written"


def run_demo():
    projection = Projection()
    lines = [f"先到版本2: {projection.receive(2, 'completed')}, 当前={projection.status}"]
    result = projection.receive(1, "running")
    lines.append(f"后到版本1: {result}, 版本={projection.version}, 当前={projection.status}")
    lines.append(f"重复版本2: {projection.receive(2, 'completed')}")
    store = FencedResult()
    old = store.acquire()
    new = store.acquire()
    lines.append(f"新工作者写入: {store.write(new, '新结果')}")
    lines.append(f"旧工作者晚到: {store.write(old, '旧结果')}, 保存={store.value}")
    return lines


if __name__ == "__main__":
    print("\n".join(run_demo()))

完整固定输出如下：

先到版本2: buffered, 当前=new
后到版本1: applied, 版本=2, 当前=completed
重复版本2: duplicate
新工作者写入: written
旧工作者晚到: stale_worker, 保存=新结果

先到的消息为什么没有立即改状态

第一行的 buffered 表示版本2暂存，当前状态仍是 new。第二行收到版本1后，while 循环发现1和2都连续，依次应用后最终为 completed。第三行相同版本、相同内容得到 duplicate，不会回退。

pending 保存尚未连续的消息，seen 保存已处理内容。如果同一个版本带不同状态，程序抛出冲突异常；仅看到版本号相同就忽略，会掩盖两个发送方写了互相矛盾的记录。版本必须由每条任务流的可靠写入端分配。

部分失败为什么产生两个工作者

调度器看到工作者1失联，不能直接证明它停止了。工作者1可能仍在计算，只是心跳无法送达；接手的工作者2完成后，工作者1又恢复提交。如果存储端接受所有结果，旧结果就会覆盖新结果。

FencedResult.acquire 递增令牌，新工作者持有2，旧工作者仍持有1。write 在写入位置核对当前令牌，因此最后一行拒绝旧结果并保留“新结果”。只在工作者开始时检查锁不够，失去所有权后它仍然可能继续运行。

版本号、幂等键与栅栏各解决什么

字段	判断内容	示例用途
消息版本	同一状态流的先后顺序	完成消息等待运行消息
业务幂等键	两次调用是否同一业务意图	重投递不再扣款
栅栏令牌	当前写入者是否仍有所有权	拒绝旧工作者覆盖

它们不应随意互换。版本2的消息重复投递仍要去重；持有最新令牌也不能重复扣款。外部服务若不接受栅栏检查，本地结果存储拒绝旧写入也不能撤销旧工作者已发出的外部动作。

缺失事件一直不来怎么办

接收端需要对缺漏设置等待时限和缓存容量，超过时限请求补发或读取权威状态。随意跳过版本缺口可能遗漏审批或失败事实。如果事件只是一份完整状态快照，也可以采用较新版本覆盖旧快照，但该契约不同于逐事件连续应用，应明确选择。

示例所有类在一个进程中，没有真实网络故障；固定到达顺序让乱序处理和栅栏判断可观察。生产写入需要由数据库条件更新或支持栅栏的存储服务原子完成令牌核对与写入。测试覆盖乱序、重复、冲突和旧执行者；Temporal 源码可用于进一步追踪跨进程的任务状态管理。

改一处输入，核对机制变化

将 run_demo 最后一次写入的 store.write(old, '旧结果') 改为 store.write(new, '第二份新结果')，运行 python -X utf8 distributed_demo.py。最后一行变为written并保存第二份新结果。令牌判断依据调用参数，打印标签不参与判断；恢复原调用再重跑，旧令牌再次被拒绝。

运行配套测试

将下面测试保存为同一文件夹的 test_distributed_demo.py。它使用固定场景检查状态与业务结果，可以直接运行：

python -X utf8 -m unittest -v test_distributed_demo.py

完整测试：test_distributed_demo.py

import unittest
from distributed_demo import Projection, FencedResult


class DistributedTests(unittest.TestCase):
    def test_out_of_order_is_buffered(self):
        state = Projection()
        self.assertEqual(state.receive(3, "completed"), "buffered")
        state.receive(1, "running")
        self.assertEqual(state.version, 1)
        state.receive(2, "reviewed")
        self.assertEqual((state.version, state.status), (3, "completed"))

    def test_duplicate_and_conflict(self):
        state = Projection()
        state.receive(1, "running")
        self.assertEqual(state.receive(1, "running"), "duplicate")
        with self.assertRaises(ValueError):
            state.receive(1, "failed")
        state.receive(3, "completed")
        with self.assertRaises(ValueError):
            state.receive(3, "failed")

    def test_stale_worker_cannot_overwrite(self):
        # 新工作者提交后，旧令牌不能覆盖已经保存的新结果。
        store = FencedResult()
        old, new = store.acquire(), store.acquire()
        self.assertEqual(store.write(new, "new"), "written")
        self.assertEqual(store.write(old, "old"), "stale_worker")
        self.assertEqual(store.value, "new")


if __name__ == "__main__":
    unittest.main()

课后练习展开习题收起习题

正在读取本地记录…

不调用模型，不执行代码。简答、代码和场景题仅保存答案并展示自查标准，不生成评分、徽章或认证。输入内容不会自动保存，请点击提交；不要填写个人信息或密钥。

练习 1 / 4 · v1.0.1

选择题 · 先判断边界

关于“乱序、部分失败与一致性”，哪种做法更稳妥？

对应知识点：L4-distributed

选择一个答案A. 先明确输入、证据和停止条件B. 默认输出永远正确C. 用一次成功推断全部场景

练习 2 / 4 · v1.0.0

简答题 · 解释失败

写出一个失败输入并说明记录字段。

对应知识点：L4-distributed

你的回答

练习 3 / 4 · v1.0.0

代码题 · 边界测试

补充正常、边界、失败三组输入和预期结果。

对应知识点：L4-distributed

你的代码或伪代码

练习 4 / 4 · v1.0.0

场景题 · 结果未知

证据不足时如何设计状态和人工接管？

对应知识点：L4-distributed

你的回答