1
-- MySQL 异步复制（默认）
2
-- 主节点写入后立即返回客户端，不等从节点
3
CHANGE MASTER TO MASTER_HOST='leader', MASTER_PORT=3306;
4
START SLAVE;
5
-- 优点：写入延迟低
6
-- 缺点：主节点故障可能丢失数据
7

8
-- MySQL 半同步复制（至少一个从节点确认）
9
-- 主节点安装半同步插件
10
INSTALL PLUGIN rpl_semi_sync_master SONAME 'semisync_master.so';
11
SET GLOBAL rpl_semi_sync_master_enabled = 1;
12
SET GLOBAL rpl_semi_sync_master_timeout = 1000; -- 等待超时 1s，超时后降级为异步
13

14
-- 从节点安装半同步插件
15
INSTALL PLUGIN rpl_semi_sync_slave SONAME 'semisync_slave.so';
16
SET GLOBAL rpl_semi_sync_slave_enabled = 1;
17
-- 优点：至少一个从节点有数据，故障转移更安全
18
-- 缺点：写入延迟增加一个 RTT

1
# 用 Python 模拟三种复制策略的行为差异
2
import time
3
import random
4

5
class ReplicationSimulator:
6
    """模拟单主复制中三种策略的延迟与数据安全差异"""
7

8
    def __init__(self, follower_count=3, network_latency_ms=50):
9
        self.follower_count = follower_count
10
        self.network_latency = network_latency_ms
11

12
    def sync_replication(self):
13
        """同步复制：等待所有从节点确认"""
14
        start = time.time()
15
        # 模拟等待所有从节点
16
        max_latency = max(
17
            random.gauss(self.network_latency, 10)
18
            for _ in range(self.follower_count)
19
        )
20
        time.sleep(max_latency / 1000)
21
        return {
22
            "strategy": "同步",
23
            "latency_ms": round(max_latency, 1),
24
            "data_safety": "所有从节点已确认，零数据丢失",
25
        }
26

27
    def async_replication(self):
28
        """异步复制：主节点写入后立即返回"""
29
        start = time.time()
30
        time.sleep(0.001)  # 仅本地写入延迟
31
        return {
32
            "strategy": "异步",
33
            "latency_ms": 1.0,
34
            "data_safety": "从节点可能未收到，主节点故障可丢数据",
35
        }
36

37
    def semi_sync_replication(self):
38
        """半同步复制：等待至少一个从节点确认"""
39
        start = time.time()
40
        min_latency = min(
41
            random.gauss(self.network_latency, 10)
42
            for _ in range(self.follower_count)
43
        )
44
        time.sleep(min_latency / 1000)
45
        return {
46
            "strategy": "半同步",
47
            "latency_ms": round(min_latency, 1),
48
            "data_safety": "至少一个从节点已确认，至多丢一个节点数据",
49
        }

1
-- MySQL Binlog 格式对比
2
-- 基于语句的日志（Statement-based）
3
-- Binlog: UPDATE users SET last_login = NOW() WHERE city = 'Beijing'
4
-- 问题：NOW() 在从节点执行时值不同！
5

6
-- 基于行的日志（Row-based）
7
-- Binlog: Table users, Row id=1: last_login='2026-05-03 10:00:00'
8
--         Table users, Row id=5: last_login='2026-05-03 10:00:01'
9
-- 优点：确定性，从节点结果与主节点完全一致
10

11
-- 设置 MySQL 使用 Row-based binlog
12
SET GLOBAL binlog_format = ROW;

1
# 模拟多主复制中的写冲突
2
# 场景：用户在两个数据中心同时修改同一用户的姓名
3

4
# 数据中心 1（北京）：用户将姓名改为 "张三"
5
dc1_data = {"user_id": 1, "name": "张三", "version": 1}
6

7
# 数据中心 2（上海）：用户将姓名改为 "张三丰"
8
dc2_data = {"user_id": 1, "name": "张三丰", "version": 1}
9

10
# 当两个数据中心的变更异步同步到对方时，冲突发生了
11
# 两个版本都基于 version=1，但值不同
12
# 这不是并发控制能解决的问题——因为两个写入发生在不同节点上
13
# 事务隔离也无法帮助——这里没有"事务"的概念

1
# LWW 的时钟偏差问题
2
# 假设两个节点的时钟有 100ms 偏差
3

4
import time
5

6
# 节点 A（时钟快 100ms）
7
node_a_time = time.time()  # 例如 1714723200.100
8
node_a_write = {"key": "x", "value": "A", "timestamp": node_a_time}
9

10
# 节点 B（时钟正常）
11
time.sleep(0.05)  # B 的写入在 A 之后 50ms 发生
12
node_b_time = time.time()  # 例如 1714723200.050
13
node_b_write = {"key": "x", "value": "B", "timestamp": node_b_time}
14

15
# LWW 判定：A 胜出（时间戳更大）
16
# 但实际上 B 的写入在 A 之后！
17
# 结果：后写入的值被先写入的值覆盖
18

19
winner = max([node_a_write, node_b_write], key=lambda x: x["timestamp"])
20
print(f"LWW winner: {winner['value']}")  # 输出: A — 错误！

1
-- Cassandra 中的 LWW 实现
2
-- Cassandra 使用客户端提供的时间戳（微秒精度）解决冲突
3
-- 后插入的行（时间戳更大）覆盖先插入的行
4

5
-- 客户端 A 写入（时间戳较大）
6
INSERT INTO users (id, name, ts)
7
VALUES (1, '张三', 1714723200100000);
8

9
-- 客户端 B 写入（时间戳较小，但实际发生更晚）
10
INSERT INTO users (id, name, ts)
11
VALUES (1, '张三丰', 1714723200050000);
12

13
-- 查询结果：name='张三'（时间戳更大者胜出）
14
-- 即使 B 的写入在逻辑上更晚，LWW 也会选择 A
15
SELECT name FROM users WHERE id = 1;

1
# Quorum 读写模拟
2
class QuorumSystem:
3
    """模拟无主复制的 Quorum 读写"""
4

5
    def __init__(self, n=3):
6
        self.n = n  # 副本总数
7
        self.nodes = [None] * n  # 每个节点存储的值
8
        self.timestamps = [0] * n  # 每个节点的写入时间戳
9

10
    def write(self, key, value, w=None):
11
        """写入：需要 w 个节点确认"""
12
        if w is None:
13
            w = self.n  # 默认写入所有节点
14
        ts = max(self.timestamps) + 1  # 递增时间戳
15
        success_count = 0
16
        for i in range(self.n):
17
            self.nodes[i] = value
18
            self.timestamps[i] = ts
19
            success_count += 1
20
            if success_count >= w:
21
                break
22
        return success_count >= w
23

24
    def read(self, key, r=None):
25
        """读取：从 r 个节点中取最新值"""
26
        if r is None:
27
            r = self.n
28
        # 收集 r 个节点的值
29
        values = []
30
        for i in range(min(r, self.n)):
31
            values.append((self.timestamps[i], self.nodes[i]))
32
        # 返回时间戳最大的值
33
        if not values:
34
            return None
35
        latest = max(values, key=lambda x: x[0])
36
        return latest[1]
37

38

39
# 常见 Quorum 配置对比
40
configs = [
41
    {"n": 3, "w": 2, "r": 2, "name": "严格 Quorum", "desc": "w+r>n，保证读到最新"},
42
    {"n": 3, "w": 1, "r": 1, "name": "宽松 Quorum", "desc": "w+r≤n，可能读到旧值"},
43
    {"n": 5, "w": 3, "r": 3, "name": "高可用 Quorum", "desc": "容忍 2 节点故障"},
44
    {"n": 5, "w": 3, "r": 2, "name": "读优化 Quorum", "desc": "写严格，读宽松"},
45
]

1
# 读修复机制模拟
2
def read_repair(quorum_system, key, r):
3
    """读取时检测并修复不一致"""
4
    # 从 r 个节点收集值
5
    responses = []
6
    for i in range(r):
7
        responses.append({
8
            "node": i,
9
            "value": quorum_system.nodes[i],
10
            "timestamp": quorum_system.timestamps[i],
11
        })
12

13
    # 找到最新值
14
    latest = max(responses, key=lambda x: x["timestamp"])
15

16
    # 修复落后节点
17
    repairs = []
18
    for resp in responses:
19
        if resp["timestamp"] < latest["timestamp"]:
20
            # 将最新值写入落后节点
21
            quorum_system.nodes[resp["node"]] = latest["value"]
22
            quorum_system.timestamps[resp["node"]] = latest["timestamp"]
23
            repairs.append(resp["node"])
24

25
    return latest["value"], repairs

1
# Cassandra 中的反熵修复命令
2
# 手动触发全量修复
3
nodetool repair --full
4

5
# 增量修复（只修复未修复的数据）
6
nodetool repair
7

8
# 查看修复进度
9
nodetool netstats

1
# 策略 1：关键读走主节点
2
# 对于用户自己修改的数据，读请求路由到主节点
3
def read_user_profile(user_id, is_self):
4
    """is_self=True 表示读取自己的资料（需要读写一致性）"""
5
    if is_self:
6
        return query_leader(f"SELECT * FROM users WHERE id = {user_id}")
7
    else:
8
        return query_follower(f"SELECT * FROM users WHERE id = {user_id}")
9

10

11
# 策略 2：时间窗口内读主节点
12
# 记录用户最后一次写入的时间，在时间窗口内读主节点
13
import time
14

15
class ReadAfterWriteRouter:
16
    def __init__(self, window_seconds=30):
17
        self.last_write_time = {}  # user_id -> timestamp
18
        self.window = window_seconds
19

20
    def after_write(self, user_id):
21
        """用户写入后，记录时间"""
22
        self.last_write_time[user_id] = time.time()
23

24
    def route_read(self, user_id):
25
        """决定读请求路由到主节点还是从节点"""
26
        last_write = self.last_write_time.get(user_id, 0)
27
        if time.time() - last_write < self.window:
28
            return "leader"  # 时间窗口内，走主节点
29
        return "follower"  # 超出窗口，走从节点

1
-- 策略 3：MySQL 中使用 GTID 追踪复制进度
2
-- 客户端写入后记录 GTID，读取时等待从节点追上
3

4
-- 主节点写入后获取 GTID
5
INSERT INTO users (name, email) VALUES ('张三', 'zhang@example.com');
6
SHOW MASTER STATUS;  -- 获取当前 GTID: 3E11FA47-71CA-11E1-9E33-C80AA9429562:1-5
7

8
-- 从节点等待 GTID
9
-- MySQL 5.6+ 支持 WAIT_FOR_EXECUTED_GTID_SET
10
SELECT WAIT_FOR_EXECUTED_GTID_SET(
11
    '3E11FA47-71CA-11E1-9E33-C80AA9429562:1-5',
12
    5  -- 最多等待 5 秒
13
);
14
-- 返回 0 表示从节点已追上，可以安全读取
15
-- 返回 1 表示超时，从节点尚未追上

1
# 单调读违反的场景模拟
2
# 用户连续两次查询，被路由到不同从节点
3

4
def query_with_monotonic_read(user_id, router, last_seen_timestamp):
5
    """保证单调读：只从时间戳 >= last_seen_timestamp 的从节点读取"""
6
    # 获取所有从节点的复制进度
7
    followers = router.get_follower_status()
8

9
    # 筛选已追上 last_seen_timestamp 的从节点
10
    eligible = [
11
        f for f in followers
12
        if f.replication_lag_ms <= last_seen_timestamp
13
    ]
14

15
    if not eligible:
16
        # 没有从节点追上，回退到主节点
17
        return router.query_leader(f"SELECT * FROM posts WHERE user_id = {user_id}")
18

19
    # 从合格的从节点中随机选择（负载均衡）
20
    import random
21
    target = random.choice(eligible)
22
    result = target.query(f"SELECT * FROM posts WHERE user_id = {user_id}")
23

24
    # 更新已见时间戳
25
    last_seen_timestamp = max(last_seen_timestamp, target.current_timestamp)
26
    return result, last_seen_timestamp

1
# 前缀一致读违反的典型场景
2
# 两个用户在聊天室对话
3

4
# 事件序列（因果顺序）：
5
# 1. Alice: "今天天气真好！"
6
# 2. Bob:   "是啊，适合出去散步。"
7

8
# 在多主复制下，Bob 的回复可能先到达某个从节点
9
# 导致读者看到：
10
# Bob: "是啊，适合出去散步。"   ← 先看到回复
11
# Alice: "今天天气真好！"       ← 后看到问题
12
# 这就是前缀一致读违反
13

14
# 解决方案：使用因果依赖追踪
15
class CausalTracker:
16
    """追踪写入间的因果关系"""
17

18
    def __init__(self):
19
        self.version_vector = {}  # node_id -> counter
20

21
    def write(self, node_id, data, dependencies=None):
22
        """写入时记录因果依赖"""
23
        if dependencies is None:
24
            dependencies = dict(self.version_vector)
25

26
        self.version_vector[node_id] = self.version_vector.get(node_id, 0) + 1
27

28
        return {
29
            "data": data,
30
            "version": dict(self.version_vector),
31
            "dependencies": dependencies,
32
        }
33

34
    def is_readable(self, write, current_version):
35
        """判断写入是否可读（所有依赖是否已满足）"""
36
        for node, counter in write["dependencies"].items():
37
            if current_version.get(node, 0) < counter:
38
                return False  # 依赖未满足，不能读
39
        return True

1
# 模拟不同拓扑下的消息传播延迟
2
import networkx as nx
3

4
def simulate_topology(topology_type, node_count=5):
5
    """模拟不同复制拓扑的消息传播"""
6
    if topology_type == "circular":
7
        # 环形拓扑：每个节点只向下一个节点传播
8
        G = nx.cycle_graph(node_count)
9
    elif topology_type == "star":
10
        # 星形拓扑：中心节点转发
11
        G = nx.star_graph(node_count - 1)
12
    elif topology_type == "all_to_all":
13
        # 全连接：每个节点直接连接所有其他节点
14
        G = nx.complete_graph(node_count)
15

16
    # 计算从节点 0 到所有其他节点的最短路径
17
    # 代表变更传播的最少跳数
18
    path_lengths = nx.single_source_shortest_path_length(G, 0)
19
    max_hops = max(path_lengths.values())
20
    avg_hops = sum(path_lengths.values()) / len(path_lengths)
21

22
    return {
23
        "topology": topology_type,
24
        "max_hops": max_hops,
25
        "avg_hops": round(avg_hops, 1),
26
        "edge_count": G.number_of_edges(),
27
    }
28

29
# 对比三种拓扑
30
for topo in ["circular", "star", "all_to_all"]:
31
    result = simulate_topology(topo, node_count=5)
32
    print(f"{result['topology']:12s} | "
33
          f"最大跳数: {result['max_hops']} | "
34
          f"平均跳数: {result['avg_hops']} | "
35
          f"边数: {result['edge_count']}")