数据库（六）：复制与分片——突破单机限制的扩展之道

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

                    Writes
Client ────────────► Leader (Primary)
                       │
              ┌────────┼────────┐
              ▼        ▼        ▼
          Follower  Follower  Follower
          (Replica) (Replica) (Replica)
              ▲        ▲        ▲
              └────────┼────────┘
                    Reads

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

-- PostgreSQL：配置同步复制
-- postgresql.conf on primary
synchronous_standby_names = 'FIRST 1 (replica1, replica2)'
-- FIRST 1 = 等待列表中至少 1 个副本确认
-- 即写入将阻塞，直到 replica1 或 replica2 中任意一个确认

-- 检查复制状态
SELECT
    client_addr,
    state,
    sync_state,
    sent_lsn,
    write_lsn,
    flush_lsn,
    replay_lsn,
    (sent_lsn - replay_lsn) AS replication_lag_bytes
FROM pg_stat_replication;

1
2
3
4
5
6

时间线：
1. 用户发布一条评论（请求发往主节点）
2. 主节点写入成功：OK
3. 用户刷新页面（请求发往从节点）
4. 该从节点尚未收到第 2 步的写入
5. 用户看到：“暂无评论”——自己的评论“消失”了！

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

# 方案 1：对近期写入的数据，强制从主节点读取
def get_user_profile(user_id, requesting_user_id):
    if user_id == requesting_user_id:
        # 用户查看自己资料 → 从主节点读
        return db_leader.query("SELECT * FROM users WHERE id = %s", user_id)
    else:
        # 查看他人资料 → 从副本读即可
        return db_replica.query("SELECT * FROM users WHERE id = %s", user_id)

# 方案 2：记录写入时间戳
def get_comments(post_id, last_write_ts=None):
    if last_write_ts and (time.time() - last_write_ts) < 5:
        # 5 秒内写入 → 从主节点读
        return db_leader.query("SELECT * FROM comments WHERE post_id = %s", post_id)
    return db_replica.query("SELECT * FROM comments WHERE post_id = %s", post_id)

1
2
3
4
5
6
7

数据中心 A              数据中心 B
┌──────────────┐          ┌──────────────┐
│   主节点 A   │◄────────►│   主节点 B   │
│  （读/写）   │          │  （读/写）   │
│      │       │          │      │       │
│  从节点      │          │  从节点      │
└──────────────┘          └──────────────┘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

-- 示例：基于时间戳的 LWW 策略
-- 两个主节点均可独立接受更新
-- 同步时，`updated_at` 时间戳最新的行获胜

-- 主节点 A：用户更新姓名
UPDATE users SET name = 'Alice Chen', updated_at = '2023-12-15T10:00:01Z'
WHERE user_id = 1;

-- 主节点 B：同一用户几乎同时更新姓名
UPDATE users SET name = 'Alice C.', updated_at = '2023-12-15T10:00:02Z'
WHERE user_id = 1;

-- 同步完成后：主节点 B 的更新胜出（时间戳更新）
-- 但主节点 A 的变更被静默丢弃

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

N = 3（每份数据有 3 个副本）
W = 2（写入需获 2 个节点确认）
R = 2（读取需从 2 个节点获取响应）

向 key "account:1" 写入 "balance = 500"：
  节点 1：✓ 已确认（balance = 500）
  节点 2：✓ 已确认（balance = 500）
  节点 3：✗ 不可达（仍为 balance = 1000）
  --> 写入成功（满足 W=2）

读取 key "account:1"：
  节点 1：balance = 500（版本 2）
  节点 2：balance = 500（版本 2）
  --> 返回 500（最新版本）
  或
  节点 2：balance = 500（版本 2）
  节点 3：balance = 1000（版本 1）
  --> 返回 500（客户端选取最高版本）

1
2
3
4
5

读取 key "account:1"：
  节点 1：balance = 500（版本 2）✓ 最新
  节点 3：balance = 1000（版本 1）✗ 陈旧
  --> 向客户端返回 500
  --> 后台：将 balance=500 写回节点 3（读修复）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

CREATE TABLE products (
    product_id INT PRIMARY KEY,
    name VARCHAR(200),
    price DECIMAL(10,2),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- 冲突时保留较新的写入
INSERT INTO products (product_id, name, price, updated_at)
VALUES (42, 'Widget', 19.99, '2024-03-15 10:30:00+00')
ON CONFLICT (product_id)
DO UPDATE SET
    name = EXCLUDED.name,
    price = EXCLUDED.price,
    updated_at = EXCLUDED.updated_at
WHERE products.updated_at < EXCLUDED.updated_at;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

class GCounter:
    """G-Counter：每个节点维护自己的计数器。"""
    def __init__(self, node_id: str):
        self.node_id = node_id
        self.counts: dict[str, int] = {}

    def increment(self):
        self.counts[self.node_id] = self.counts.get(self.node_id, 0) + 1

    def value(self) -> int:
        return sum(self.counts.values())

    def merge(self, other: "GCounter"):
        """合并满足交换律、结合律、幂等性。"""
        for node, count in other.counts.items():
            self.counts[node] = max(self.counts.get(node, 0), count)

1
2
3
4
5
6
7
8
9

CREATE TABLE document_versions (
    doc_id INT,
    version_id UUID DEFAULT gen_random_uuid(),
    content TEXT,
    author VARCHAR(50),
    written_at TIMESTAMPTZ,
    is_conflict BOOLEAN DEFAULT FALSE,
    PRIMARY KEY (doc_id, version_id)
);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

import random

class DatabaseRouter:
    def __init__(self, primary: str, replicas: list[str]):
        self.primary = primary
        self.replicas = replicas

    def get_connection(self, operation: str, consistency: str = "eventual"):
        if operation in ("INSERT", "UPDATE", "DELETE"):
            return self.primary
        if consistency == "strong":
            return self.primary
        return random.choice(self.replicas)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

import time

class SessionConsistency:
    """保证每个会话的读自己写。"""

    def __init__(self, router: DatabaseRouter):
        self.router = router
        self.last_write_time: float = 0
        self.lag_threshold: float = 2.0  # 秒

    def get_read_connection(self):
        if time.time() - self.last_write_time < self.lag_threshold:
            return self.router.primary  # 太近，用主库
        return self.router.get_connection("SELECT")

    def record_write(self):
        self.last_write_time = time.time()

1
2
3
4
5
6
7
8
9

-- PostgreSQL：检查复制延迟
SELECT
    client_addr, state,
    pg_wal_lsn_diff(sent_lsn, replay_lsn) AS replay_lag_bytes
FROM pg_stat_replication;

-- MySQL：检查落后秒数
SHOW SLAVE STATUS\G
-- 关注: Seconds_Behind_Master

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

# patroni.yml
scope: my-cluster
name: node1

etcd3:
  hosts: etcd1:2379,etcd2:2379,etcd3:2379

bootstrap:
  dcs:
    ttl: 30
    loop_wait: 10
    maximum_lag_on_failover: 1048576  # 提升时最大允许 1MB 延迟

postgresql:
  listen: 0.0.0.0:5432
  parameters:
    max_connections: 200
    synchronous_commit: "on"

1
2
3

分片 1：user_id    1 - 1,000,000
分片 2：user_id    1,000,001 - 2,000,000
分片 3：user_id    2,000,001 - 3,000,000

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

-- PostgreSQL 声明式分区（按范围）
CREATE TABLE orders (
    order_id    BIGSERIAL,
    user_id     INT NOT NULL,
    created_at  TIMESTAMP NOT NULL,
    total       DECIMAL(10,2)
) PARTITION BY RANGE (created_at);

CREATE TABLE orders_2023_q4 PARTITION OF orders
    FOR VALUES FROM ('2023-10-01') TO ('2024-01-01');

CREATE TABLE orders_2024_q1 PARTITION OF orders
    FOR VALUES FROM ('2024-01-01') TO ('2024-04-01');

CREATE TABLE orders_2024_q2 PARTITION OF orders
    FOR VALUES FROM ('2024-04-01') TO ('2024-07-01');

-- 查询自动路由至正确分区
SELECT * FROM orders WHERE created_at = '2023-11-15';
-- 仅扫描 orders_2023_q4，跳过其他分区

1
2
3
4
5

分片编号 = hash(user_id) % 分片总数

hash("user:1")  = 0x3A2B... → 分片 2
hash("user:2")  = 0x8F1C... → 分片 0
hash("user:3")  = 0x12D4... → 分片 1

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

-- PostgreSQL 哈希分区
CREATE TABLE sessions (
    session_id  UUID PRIMARY KEY,
    user_id     INT NOT NULL,
    data        JSONB,
    expires_at  TIMESTAMP
) PARTITION BY HASH (session_id);

CREATE TABLE sessions_p0 PARTITION OF sessions
    FOR VALUES WITH (MODULUS 4, REMAINDER 0);
CREATE TABLE sessions_p1 PARTITION OF sessions
    FOR VALUES WITH (MODULUS 4, REMAINDER 1);
CREATE TABLE sessions_p2 PARTITION OF sessions
    FOR VALUES WITH (MODULUS 4, REMAINDER 2);
CREATE TABLE sessions_p3 PARTITION OF sessions
    FOR VALUES WITH (MODULUS 4, REMAINDER 3);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

                    0 / 2^32
                      │
              节点 C  ●
                     ╱  ╲
                    ╱    ╲
           ●──────╱──────●
         节点 A  ╱      节点 B
                ╱
               ╱
键 "user:42" 哈希至位置 X
→ 顺时针遍历 → 首个遇到的节点即为归属者

新增节点 D：
  仅节点 C 与 D 之间的键需迁移
  （而非全集群所有键）

1
2
3
4
5

物理节点 A → 虚拟节点：A1, A2, A3, A4, A5（环上 5 个位置）
物理节点 B → 虚拟节点：B1, B2, B3, B4, B5

虚拟节点越多 → 分布越均衡
Cassandra 默认：每物理节点 256 个 vnode

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

扩容前（10 节点，1000 分区）：
  节点 1：分区 0–99
  节点 2：分区 100–199
  ...

新增节点 11 后：
  节点 1：分区 0–89（让出 10 个）
  节点 2：分区 100–189（让出 10 个）
  ...
  节点 11：分区 90–99, 190–199, ...（共接收约 91 个分区）

1
2
3
4

-- 表按 user_id 分片
-- 但我们还需按 email 查询
SELECT * FROM users WHERE email = 'alice@example.com';
-- 这个用户在哪一分片？无法得知，只能遍历全部分片。

1
2
3
4
5

分片 1：email 本地索引 → {alice@...: row 1, bob@...: row 2}
分片 2：email 本地索引 → {carol@...: row 3, dave@...: row 4}

按 email 查询 → 向**所有分片**广播请求，聚合结果
（即 “scatter-gather” —— 扇出成本高昂）

1
2
3
4
5

Email 索引分片 A（a–m）：alice@... → 分片 1，carol@... → 分片 2
Email 索引分片 B（n–z）：zara@... → 分片 3

按 email 查询 → 先查索引分片 → 再查数据分片
（2 次跳转，但无需 scatter-gather）

1
2
3
4
5
6
7
8

# /etc/mysql/mysql.conf.d/mysqld.cnf on the leader
[mysqld]
server-id           = 1
log_bin              = /var/log/mysql/mysql-bin
binlog_format        = ROW          # 最安全格式
binlog_expire_logs_seconds = 604800 # 7 天日志保留
sync_binlog          = 1            # 每次提交均刷盘 binlog
innodb_flush_log_at_trx_commit = 1  # 完整持久化

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

-- 在主节点创建复制用户
CREATE USER 'repl_user'@'%' IDENTIFIED BY 'strong_password_here';
GRANT REPLICATION SLAVE ON *.* TO 'repl_user'@'%';
FLUSH PRIVILEGES;

-- 获取当前 binlog 位置
SHOW MASTER STATUS;
-- +------------------+----------+
-- | File             | Position |
-- +------------------+----------+
-- | mysql-bin.000003 |      785 |
-- +------------------+----------+

1
2
3
4
5
6
7
8

# 方案 1：mysqldump + 一致性快照
mysqldump --all-databases --single-transaction \
  --source-data=2 --routines --triggers \
  -u root -p > leader_backup.sql

# 方案 2：大数据量场景使用 xtrabackup
xtrabackup --backup --target-dir=/backup/full \
  --user=root --password=xxx

1
2
3
4
5
6

# /etc/mysql/mysql.conf.d/mysqld.cnf on the follower
[mysqld]
server-id            = 2       # 必须唯一
relay_log            = /var/log/mysql/mysql-relay
read_only            = ON      # 防止误写
super_read_only      = ON      # 连 root 也无法写

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

-- 在从节点恢复备份
-- 然后配置复制源
CHANGE REPLICATION SOURCE TO
    SOURCE_HOST='leader-hostname',
    SOURCE_USER='repl_user',
    SOURCE_PASSWORD='strong_password_here',
    SOURCE_LOG_FILE='mysql-bin.000003',
    SOURCE_LOG_POS=785;

-- 启动复制
START REPLICA;

-- 检查复制状态
SHOW REPLICA STATUS\G
-- 关键字段检查：
--   Replica_IO_Running: Yes
--   Replica_SQL_Running: Yes
--   Seconds_Behind_Source: 0
--   Last_Error: （应为空）

1
2
3
4
5
6
7
8
9

-- 在从节点：检查延迟
SHOW REPLICA STATUS\G
-- Seconds_Behind_Source: 0  <-- 健康
-- Seconds_Behind_Source: 45 <-- 需关注
-- Seconds_Behind_Source: NULL <-- 复制已中断！

-- 在主节点：查看已连接的从节点
SHOW REPLICAS;
-- 或旧语法：SHOW SLAVE HOSTS;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# 快速健康检查脚本
#!/bin/bash
LAG=$(mysql -e "SHOW REPLICA STATUS\G" | grep "Seconds_Behind_Source" | awk '{print $2}')
IO_RUNNING=$(mysql -e "SHOW REPLICA STATUS\G" | grep "Replica_IO_Running" | awk '{print $2}')
SQL_RUNNING=$(mysql -e "SHOW REPLICA STATUS\G" | grep "Replica_SQL_Running" | awk '{print $2}')

echo "Replication Lag: ${LAG}s"
echo "IO Thread: $IO_RUNNING"
echo "SQL Thread: $SQL_RUNNING"

if [ "$LAG" -gt 60 ] || [ "$IO_RUNNING" != "Yes" ] || [ "$SQL_RUNNING" != "Yes" ]; then
    echo "ALERT: Replication unhealthy!"
    exit 1
fi

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

-- 在待提升的从节点上执行：
STOP REPLICA;
RESET REPLICA ALL;

-- 该节点现已成为独立服务器
-- 重新配置其余从节点，使其指向新主节点

-- 在剩余从节点上执行：
STOP REPLICA;
CHANGE REPLICATION SOURCE TO
    SOURCE_HOST='new-leader-hostname',
    SOURCE_LOG_FILE='...',
    SOURCE_LOG_POS=...;
START REPLICA;

-- 应用程序配置也需更新
-- （或使用 ProxySQL / HAProxy 等代理层）