1
-- 垂直分库后：按业务域拆分
2
-- user_db: users, user_profiles, user_settings
3
-- order_db: orders, order_items, refunds
4
-- product_db: products, categories, inventory
5
-- payment_db: payments, transactions, invoices

1
# 范围分片路由逻辑
2
RANGE_RULES = [
3
    (1, 10_000_000, "shard_0"),
4
    (10_000_001, 50_000_000, "shard_1"),
5
    (50_000_001, 100_000_000, "shard_2"),
6
]
7

8
def route_by_range(key: int) -> str:
9
    for start, end, shard in RANGE_RULES:
10
        if start <= key <= end:
11
            return shard
12
    raise ValueError(f"Key {key} out of range")

1
// 哈希分片路由逻辑
2
public class HashSharding {
3
    private final int shardCount;
4

5
    public String route(String shardKey) {
6
        int hash = murmurHash3(shardKey);
7
        int shardIndex = Math.abs(hash) % shardCount;
8
        return "shard_" + shardIndex;
9
    }
10

11
    private int murmurHash3(String key) {
12
        int h = 0;
13
        for (int i = 0; i < key.length(); i++) {
14
            h = h * 31 + key.charAt(i);
15
            h ^= (h >>> 16);
16
        }
17
        return h;
18
    }
19
}

1
# 一致性哈希分片实现
2
import hashlib
3
from bisect import bisect
4

5
class ConsistentHashing:
6
    def __init__(self, replicas=150):
7
        self.replicas = replicas
8
        self.ring = []
9
        self.node_map = {}
10

11
    def _hash(self, key: str) -> int:
12
        return int(hashlib.md5(key.encode()).hexdigest(), 16)
13

14
    def add_node(self, node: str):
15
        for i in range(self.replicas):
16
            h = self._hash(f"{node}#{i}")
17
            self.ring.append(h)
18
            self.node_map[h] = node
19
        self.ring.sort()
20

21
    def route(self, key: str) -> str:
22
        h = self._hash(key)
23
        idx = bisect(self.ring, h)
24
        if idx == len(self.ring):
25
            idx = 0
26
        return self.node_map[self.ring[idx]]

1
-- 分片映射表
2
CREATE TABLE shard_mapping (
3
    entity_type VARCHAR(50) NOT NULL,
4
    entity_id   BIGINT NOT NULL,
5
    shard_id    INT NOT NULL,
6
    PRIMARY KEY (entity_type, entity_id)
7
);
8

9
-- 路由查询：先查映射表，再查目标分片
10
SELECT shard_id FROM shard_mapping
11
WHERE entity_type = 'order' AND entity_id = 12345;

1
UUID.randomUUID().toString();
2
// 输出: 550e8400-e29b-41d4-a716-446655440000

1
/**
2
 * Snowflake ID 结构（64 位）：
3
 * | 1 bit | 41 bits      | 10 bits  | 12 bits  |
4
 * | 符号位 | 时间戳（毫秒） | 机器 ID  | 序列号    |
5
 * - 41 位时间戳：约 69 年
6
 * - 10 位机器 ID：1024 台机器
7
 * - 12 位序列号：每毫秒 4096 个 ID
8
 */
9
public class SnowflakeIdGenerator {
10
    private final long epoch = 1609459200000L;
11
    private final long sequenceMask = ~(-1L << 12L);
12
    private final long workerIdShift = 12L;
13
    private final long datacenterIdShift = 17L;
14
    private final long timestampShift = 22L;
15
    private final long workerId;
16
    private final long datacenterId;
17
    private long sequence = 0L;
18
    private long lastTimestamp = -1L;
19

20
    public synchronized long nextId() {
21
        long timestamp = System.currentTimeMillis();
22
        if (timestamp < lastTimestamp)
23
            throw new RuntimeException("时钟回拨，拒绝生成 ID");
24
        if (timestamp == lastTimestamp) {
25
            sequence = (sequence + 1) & sequenceMask;
26
            if (sequence == 0) timestamp = tilNextMillis(lastTimestamp);
27
        } else {
28
            sequence = 0L;
29
        }
30
        lastTimestamp = timestamp;
31
        return ((timestamp - epoch) << timestampShift)
32
             | (datacenterId << datacenterIdShift)
33
             | (workerId << workerIdShift)
34
             | sequence;
35
    }
36

37
    // tilNextMillis: 自旋等待下一毫秒
38
}

1
-- 号段表设计
2
CREATE TABLE id_segment (
3
    biz_tag    VARCHAR(64) PRIMARY KEY,
4
    max_id     BIGINT NOT NULL,
5
    step       INT NOT NULL,
6
    version    INT NOT NULL,
7
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
8
);
9

10
-- 获取号段（乐观锁更新）
11
UPDATE id_segment
12
SET max_id = max_id + step, version = version + 1
13
WHERE biz_tag = 'order' AND version = 0;
14
-- 成功后，本地拥有 [1, 10000] 的 ID 段

1
-- 单库：简单的用户-订单 Join
2
SELECT u.name, o.order_no, o.amount
3
FROM users u JOIN orders o ON u.id = o.user_id
4
WHERE u.city = '北京';
5

6
-- 分库后：应用层 Join（推荐）
7
-- Step 1: 查询北京用户
8
SELECT id, name FROM user_db.users WHERE city = '北京';
9
-- Step 2: 根据用户 ID 查询订单
10
SELECT user_id, order_no, amount
11
FROM order_db.orders WHERE user_id IN (1001, 1002, 1003);

1
# 跨分片聚合：两阶段处理（分片内聚合 + 合并）
2
def sharded_count(table, where):
3
    results = [execute(shard, f"SELECT COUNT(*) AS c FROM {table} WHERE {where}")
4
               for shard in shards]
5
    return sum(r[0]['c'] for r in results)  # 各分片 COUNT 求和
6

7
def sharded_avg(table, col, where):
8
    results = [execute(shard, f"SELECT SUM({col}) AS s, COUNT({col}) AS c FROM {table} WHERE {where}")
9
               for shard in shards]
10
    return sum(r['s'] for r in results) / sum(r['c'] for r in results)

1
-- 错误：每个分片 LIMIT 10, 5 然后合并 → 结果不正确
2
-- 正确：各分片取 top N，内存归并排序
3
-- 分片1: SELECT * FROM orders ORDER BY created_at LIMIT 0, 15
4
-- 分片2: SELECT * FROM orders ORDER BY created_at LIMIT 0, 15
5
-- 内存归并排序后取第 10~15 条

1
// XA 事务示例（Java + Atomikos）
2
UserTransaction tx = utm.getTransaction();
3
tx.begin();
4
// 操作分片1：扣款
5
Connection conn1 = getShardConnection(fromUserId);
6
conn1.createStatement().executeUpdate(
7
    "UPDATE accounts SET balance = balance - " + amount
8
    + " WHERE user_id = " + fromUserId);
9
// 操作分片2：加款
10
Connection conn2 = getShardConnection(toUserId);
11
conn2.createStatement().executeUpdate(
12
    "UPDATE accounts SET balance = balance + " + amount
13
    + " WHERE user_id = " + toUserId);
14
tx.commit();  // 两阶段提交：prepare → commit/rollback

1
# Saga 模式：电商下单流程
2
class OrderSaga:
3
    def execute(self, order):
4
        saga = SagaBuilder() \
5
            .step("创建订单",
6
                  action=lambda: self.order_service.create(order),
7
                  compensate=lambda: self.order_service.cancel(order.id)) \
8
            .step("扣减库存",
9
                  action=lambda: self.inventory_service.deduct(order.sku_id, order.qty),
10
                  compensate=lambda: self.inventory_service.restore(order.sku_id, order.qty)) \
11
            .step("扣减余额",
12
                  action=lambda: self.account_service.deduct(order.user_id, order.amount),
13
                  compensate=lambda: self.account_service.refund(order.user_id, order.amount)) \
14
            .build()
15
        saga.run()  # 任何步骤失败，自动逆序执行补偿

1
-- 本地消息表
2
CREATE TABLE outbox_messages (
3
    id          BIGINT PRIMARY KEY AUTO_INCREMENT,
4
    biz_type    VARCHAR(64) NOT NULL,
5
    biz_id      VARCHAR(128) NOT NULL,
6
    payload     TEXT NOT NULL,
7
    status      ENUM('PENDING','SENT','FAILED') DEFAULT 'PENDING',
8
    retry_count INT DEFAULT 0,
9
    max_retry   INT DEFAULT 5,
10
    INDEX idx_status_created (status, created_at)
11
);
12

13
-- 业务操作 + 消息写入在同一个本地事务中
14
BEGIN;
15
INSERT INTO orders (user_id, amount, status) VALUES (1001, 99.9, 'CREATED');
16
INSERT INTO outbox_messages (biz_type, biz_id, payload)
17
VALUES ('ORDER_CREATED', 'ORD_001',
18
        '{"order_id":"ORD_001","user_id":1001,"amount":99.9}');
19
COMMIT;

1
-- TiDB 分布式事务示例
2
BEGIN OPTIMISTIC;
3
UPDATE accounts SET balance = balance - 100 WHERE user_id = 1;
4
UPDATE accounts SET balance = balance + 100 WHERE user_id = 2;
5
COMMIT;  -- 两阶段提交：Prewrite → Commit

1
-- OceanBase：多租户资源隔离 + 兼容 MySQL 协议
2
CREATE RESOURCE UNIT unit_app1 MAX_CPU 4, MIN_CPU 2, MEMORY_SIZE '8G';
3
CREATE RESOURCE POOL pool_app1 UNIT = unit_app1, UNIT_NUM = 3;
4
CREATE TENANT app1_tenant RESOURCE_POOL_LIST = ('pool_app1'), CHARSET = UTF8;
5

6
-- 支持 Hash/Range/List 分区
7
CREATE TABLE orders (id BIGINT PRIMARY KEY, user_id BIGINT, amount DECIMAL(10,2))
8
PARTITION BY HASH(id) PARTITIONS 16;