分布式快照——1985年，为什么不暂停整个网络，也能拍下一张全局一致的照片？

transfer(100)

MARKER

from collections import deque

nodes = ["A", "B"]

balances = {
    "A": 1000,
    "B": 500,
}

# 两条独立的 FIFO 单向通道
channels = {
    ("A", "B"): deque(),
    ("B", "A"): deque(),
}

# 本地状态快照
local_snapshot = {}

# 各条通道的快照
channel_snapshot = {}

# 当前仍在记录的入站通道
recording = set()


def send(src, dst, kind, amount=0):
    channels[(src, dst)].append(
        (kind, amount)
    )


def record_local(node, first_marker_channel=None):
    """
    保存节点的本地快照。

    first_marker_channel:
    导致该节点第一次开始快照的入站通道。
    这条通道的快照状态为空。
    """
    local_snapshot[node] = balances[node]

    # 初始化所有入站通道的快照
    for channel in channels:
        src, dst = channel

        if dst != node:
            continue

        channel_snapshot[channel] = []

        if channel != first_marker_channel:
            recording.add(channel)

    # 保存本地状态后，先发 Marker
    for channel in channels:
        src, dst = channel

        if src == node:
            send(src, dst, "MARKER")


def receive(src, dst):
    channel = (src, dst)

    kind, amount = channels[channel].popleft()

    if kind == "MARKER":

        if dst not in local_snapshot:
            # 第一次收到 Marker
            record_local(
                dst,
                first_marker_channel=channel,
            )

        else:
            # 已经记录本地快照；
            # 此通道现在停止记录
            recording.discard(channel)

    elif kind == "TRANSFER":

        # 如果此通道处于快照记录阶段，
        # 保存这条在途消息
        if channel in recording:
            channel_snapshot[channel].append(amount)

        # 正常业务处理仍然继续
        balances[dst] += amount

# 1. A 扣款，并发送转账消息
balances["A"] -= 100
send("A", "B", "TRANSFER", 100)

# 2. B 主动发起快照
record_local("B")

# 3. A 收到 B 发来的 Marker
receive("B", "A")

# 4. B 收到 A 之前发出的转账消息
receive("A", "B")

# 5. B 收到 A 随后发出的 Marker
receive("A", "B")

print(local_snapshot)

print(channel_snapshot)

snapshot_total = (
    sum(local_snapshot.values())
    + sum(
        sum(messages)
        for messages in channel_snapshot.values()
    )
)

print(snapshot_total)
print(balances)

{'B': 500, 'A': 900}

{('A', 'B'): [100], ('B', 'A'): []}

1500

{'A': 900, 'B': 600}

A = 900
B = 500
A→B 通道 = 100

A = 900
B = 600
A→B 通道 = 空

A：空闲
B：空闲
C：空闲

Kafka 消息
    ↓
订单过滤
    ↓
按用户分组
    ↓
实时汇总
    ↓
写入下游存储