从零实现一个简易的分库分表

最近一个老项目的订单表眼瞅着快 3000 万了，单表查询开始有点顶不住。本来想直接上 ShardingSphere，结果看了一眼配置，头有点大。干脆自己写一个简易的，先把分库分表这事儿跑通，后面再换中间件也不迟。下面是实测能跑通的版本，代码不复杂，但坑不少。

先把分片规则定下来

这里不搞太复杂，就按 user_id 分。规则是：

db_index = user_id % 2
table_index = (user_id // 2) % 2

也就是说，两个库，每个库两张表，一共 4 个分片。同一个用户的订单永远落在同一个库同一张表里，按 user_id 查询可以直接路由，不用扫全部分片。这个规则很土，但用来演示分库分表的基本流程足够了。

表名就固定为 orders_0 和 orders_1，库名分别是 sharding_demo_0 和 sharding_demo_1。下面开始建库建表。

建库建表

首先把两个库建出来：

CREATE DATABASE IF NOT EXISTS sharding_demo_0 DEFAULT CHARSET utf8mb4;
CREATE DATABASE IF NOT EXISTS sharding_demo_1 DEFAULT CHARSET utf8mb4;

然后在 sharding_demo_0 里执行：

USE sharding_demo_0;

CREATE TABLE IF NOT EXISTS orders_0 (
  id BIGINT PRIMARY KEY,
  user_id BIGINT NOT NULL,
  amount DECIMAL(10,2) NOT NULL,
  created_at DATETIME NOT NULL,
  KEY idx_user_created (user_id, created_at)
) ENGINE=InnoDB;

CREATE TABLE IF NOT EXISTS orders_1 (
  id BIGINT PRIMARY KEY,
  user_id BIGINT NOT NULL,
  amount DECIMAL(10,2) NOT NULL,
  created_at DATETIME NOT NULL,
  KEY idx_user_created (user_id, created_at)
) ENGINE=InnoDB;

接着把 USE sharding_demo_0; 换成 USE sharding_demo_1;，再把上面两张表建一遍。别偷懒，两个库都得有这两张表，不然路由到没建表的分片就直接报错了。

写路由和全局 ID

分库分表之后，数据库自增 ID 就不能用了，几个分片各增各的，迟早撞 ID。简易版我直接用 时间戳毫秒 * 1000 + 序列号，单机测试够用。多实例部署这玩意儿会重复，别直接拿去生产。

先装个依赖：

pip install pymysql

然后核心代码大概长这样：

import time
import pymysql
from decimal import Decimal

DB_CONFIG = {
    0: dict(
        host="127.0.0.1",
        port=3306,
        user="root",
        password="123456",
        database="sharding_demo_0",
        charset="utf8mb4",
        autocommit=False,
    ),
    1: dict(
        host="127.0.0.1",
        port=3306,
        user="root",
        password="123456",
        database="sharding_demo_1",
        charset="utf8mb4",
        autocommit=False,
    ),
}


class IdGen:
    def __init__(self):
        self.last_ms = 0
        self.seq = 0

    def next_id(self):
        now = int(time.time() * 1000)
        if now == self.last_ms:
            self.seq += 1
        else:
            self.last_ms = now
            self.seq = 0
        return now * 1000 + self.seq


class SimpleSharding:
    def __init__(self):
        self.conns = {i: pymysql.connect(**DB_CONFIG[i]) for i in range(2)}
        self.id_gen = IdGen()

    def route(self, user_id):
        db_index = user_id % 2
        table_index = (user_id // 2) % 2
        return db_index, f"orders_{table_index}"

    def insert_order(self, user_id, amount):
        order_id = self.id_gen.next_id()
        db_index, table = self.route(user_id)
        conn = self.conns[db_index]
        with conn.cursor() as cur:
            cur.execute(
                f"INSERT INTO {table}(id, user_id, amount, created_at) "
                f"VALUES(%s, %s, %s, NOW())",
                (order_id, user_id, amount),
            )
        conn.commit()
        return order_id

    def find_by_user(self, user_id, limit=20, offset=0):
        db_index, table = self.route(user_id)
        conn = self.conns[db_index]
        with conn.cursor(pymysql.cursors.DictCursor) as cur:
            cur.execute(
                f"SELECT * FROM {table} WHERE user_id=%s "
                f"ORDER BY created_at DESC LIMIT %s OFFSET %s",
                (user_id, limit, offset),
            )
            return cur.fetchall()

    def find_all(self, limit=20, offset=0):
        rows = []
        for db_index in range(2):
            conn = self.conns[db_index]
            with conn.cursor(pymysql.cursors.DictCursor) as cur:
                for table in ("orders_0", "orders_1"):
                    cur.execute(f"SELECT * FROM {table}")
                    rows.extend(cur.fetchall())

        rows.sort(key=lambda r: (r["created_at"], r["id"]), reverse=True)
        return rows[offset:offset + limit]

这里 route 就是整个分库分表的核心，插入和按用户查询都靠它找到对应的库和表。find_all 是全局查询，把 4 个分片全扫一遍再在内存里排序分页，数据量小的时候能跑，数据量大了就是找死，这个后面再说。

跑一下

把 DB_CONFIG 里的账号密码改成你自己的，然后直接运行：

if __name__ == "__main__":
    sharding = SimpleSharding()

    for uid in [1, 2, 3, 4, 5, 6]:
        sharding.insert_order(uid, Decimal("19.90"))

    print("用户 1 的订单：", sharding.find_by_user(1, limit=10))
    print("全局最近 10 条：", sharding.find_all(limit=10))

不出问题的话，你就能看到用户 1 的订单和全局最近 10 条了。出了问题的，大概率是库名、表名或者密码没改，先检查这几个地方。

几个坑

第一个坑是分页。像 find_all 这种把全部分片数据拉到内存再分页，只适合数据量很小的时候。真要做跨分片分页，得每个分片先查 offset + limit，再把结果归并排序，最后取需要的区间。而且 offset 越大越慢，这个到后面是另一个坑。

第二个坑是事务。现在代码里跨库事务根本没处理，单分片插入提交没问题，但你要一次插两个不同用户的数据，就不在一个事务里了。生产上要么搞本地消息表、最终一致，要么直接上靠谱的中间件，别自己硬写。

第三个坑是扩容。现在 user_id % 2 这种取模，以后想扩成 4 个库就得迁数据。要么提前把库数量规划多一点，要么用一致性哈希，但一致性哈希又会带来查询和扩容的新问题。

最后，跨分片 join 就别想了，我这里是直接没做。真需要 join 的字段，要么冗余，要么在业务层拼。

先跑通这些，后面再折腾扩容和事务，不然一开始就上全套，很容易把自己绕进去。