1
CREATE TABLE users (
2
    id          BIGINT       NOT NULL COMMENT '用户ID（Snowflake）',
3
    username    VARCHAR(50)  NOT NULL COMMENT '用户名',
4
    email       VARCHAR(255) NOT NULL COMMENT '邮箱',
5
    phone       VARCHAR(20)  DEFAULT NULL COMMENT '手机号',
6
    password_hash VARCHAR(255) NOT NULL COMMENT '密码哈希',
7
    avatar_url  VARCHAR(512) DEFAULT NULL COMMENT '头像URL',
8
    status      TINYINT      NOT NULL DEFAULT 1 COMMENT '1正常 2禁用',
9
    created_at  DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,
10
    updated_at  DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
11
    PRIMARY KEY (id),
12
    UNIQUE KEY uk_email (email),
13
    UNIQUE KEY uk_phone (phone),
14
    KEY idx_username (username),
15
    KEY idx_created_at (created_at)
16
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

1
CREATE TABLE products (
2
    id           BIGINT        NOT NULL COMMENT '商品ID',
3
    name         VARCHAR(200)  NOT NULL COMMENT '商品名称',
4
    category_id  BIGINT        NOT NULL COMMENT '分类ID',
5
    price        DECIMAL(10,2) NOT NULL COMMENT '售价',
6
    original_price DECIMAL(10,2) DEFAULT NULL COMMENT '原价',
7
    stock        INT           NOT NULL DEFAULT 0 COMMENT '库存',
8
    sales_count  INT           NOT NULL DEFAULT 0 COMMENT '销量（反范式）',
9
    status       TINYINT       NOT NULL DEFAULT 1 COMMENT '1上架 2下架',
10
    created_at   DATETIME      NOT NULL DEFAULT CURRENT_TIMESTAMP,
11
    updated_at   DATETIME      NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
12
    PRIMARY KEY (id),
13
    KEY idx_category (category_id),
14
    KEY idx_status_created (status, created_at),
15
    KEY idx_sales (sales_count DESC)
16
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

1
CREATE TABLE orders (
2
    id            BIGINT        NOT NULL COMMENT '订单ID（Snowflake）',
3
    user_id       BIGINT        NOT NULL COMMENT '买家ID',
4
    status        TINYINT       NOT NULL DEFAULT 1 COMMENT '1待付 2已付 3已发 4完成 5取消',
5
    total_amount  DECIMAL(12,2) NOT NULL COMMENT '订单总金额（反范式快照）',
6
    payment_time  DATETIME      DEFAULT NULL COMMENT '支付时间',
7
    created_at    DATETIME      NOT NULL DEFAULT CURRENT_TIMESTAMP,
8
    updated_at    DATETIME      NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
9
    PRIMARY KEY (id),
10
    KEY idx_user_created (user_id, created_at),
11
    KEY idx_status (status),
12
    KEY idx_created_at (created_at)
13
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

1
# Cache-Aside 读取
2
async def get_product(product_id: int) -> Product:
3
    # 1. 查缓存
4
    cache_key = f"product:{product_id}"
5
    cached = await redis.get(cache_key)
6
    if cached:
7
        return Product.from_json(cached)
8

9
    # 2. 缓存 miss，查数据库
10
    product = await db.query(
11
        "SELECT * FROM products WHERE id = %s", product_id
12
    )
13
    if product:
14
        # 3. 回写缓存
15
        await redis.setex(cache_key, 3600, product.to_json())
16

17
    return product
18

19
# Cache-Aside 更新
20
async def update_product(product_id: int, data: dict) -> None:
21
    # 1. 先更新数据库
22
    await db.execute(
23
        "UPDATE products SET name=%s, price=%s WHERE id=%s",
24
        data["name"], data["price"], product_id
25
    )
26
    # 2. 再删除缓存（而非更新缓存）
27
    await redis.delete(f"product:{product_id}")

1
# 延迟双删：适用于库存等关键数据
2
async def update_stock(product_id: int, delta: int) -> None:
3
    cache_key = f"stock:{product_id}"
4

5
    # 1. 先删缓存
6
    await redis.delete(cache_key)
7

8
    # 2. 更新数据库
9
    await db.execute(
10
        "UPDATE products SET stock = stock + %s WHERE id = %s AND stock + %s >= 0",
11
        delta, product_id, delta
12
    )
13

14
    # 3. 延迟再删一次（防止并发读回写旧值）
15
    await asyncio.sleep(0.5)  # 延迟 > 主从复制延迟
16
    await redis.delete(cache_key)

1
/**
2
 * Snowflake ID 结构（64 bit）：
3
 * | 1 bit 符号 | 41 bit 时间戳 | 5 bit 数据中心 | 5 bit 机器 | 12 bit 序列号 |
4
 * 每毫秒可生成 4096 个 ID，最多支持 1024 节点
5
 */
6
public class SnowflakeIdGenerator {
7
    private final long epoch = 1704067200000L; // 2024-01-01
8
    private final long datacenterId, workerId;
9
    private long sequence = 0, lastTimestamp = -1L;
10

11
    public synchronized long nextId() {
12
        long ts = System.currentTimeMillis();
13
        if (ts == lastTimestamp) {
14
            sequence = (sequence + 1) & 0xFFF;
15
            if (sequence == 0) ts = waitNextMillis(ts);
16
        } else { sequence = 0; }
17
        lastTimestamp = ts;
18
        return ((ts - epoch) << 22) | (datacenterId << 17)
19
             | (workerId << 12) | sequence;
20
    }
21
}

1
# ShardingSphere 分片规则
2
rules:
3
  - !SHARDING
4
    tables:
5
      orders:
6
        actualDataNodes: order_db_${0..15}.orders_${0..3}
7
        tableStrategy:
8
          standard:
9
            shardingColumn: user_id
10
            shardingAlgorithmName: orders_mod
11
        keyGenerateStrategy:
12
          column: id
13
          keyGeneratorName: snowflake
14

15
    shardingAlgorithms:
16
      orders_mod:
17
        type: MOD
18
        props:
19
          sharding-count: 64  # 16库 × 4表
20

21
    keyGenerators:
22
      snowflake:
23
        type: SNOWFLAKE
24
        props:
25
          worker-id: 1

1
-- 每个服务本地消息表
2
CREATE TABLE outbox_messages (
3
    id          BIGINT       NOT NULL AUTO_INCREMENT,
4
    aggregate_id BIGINT      NOT NULL COMMENT '聚合根ID（如订单ID）',
5
    event_type  VARCHAR(100) NOT NULL COMMENT '事件类型',
6
    payload     JSON         NOT NULL COMMENT '事件内容',
7
    status      TINYINT      NOT NULL DEFAULT 0 COMMENT '0待发 1已发 2失败',
8
    retry_count INT          NOT NULL DEFAULT 0,
9
    created_at  DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,
10
    PRIMARY KEY (id),
11
    KEY idx_status_created (status, created_at),
12
    KEY idx_aggregate (aggregate_id)
13
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

1
// 订单创建 + 消息投递（同一本地事务）
2
@Transactional
3
public Order createOrder(CreateOrderCmd cmd) {
4
    // 1. 创建订单
5
    Order order = Order.create(cmd);
6
    orderMapper.insert(order);
7

8
    // 2. 写入本地消息表（与订单在同一事务中）
9
    OutboxMessage msg = OutboxMessage.builder()
10
        .aggregateId(order.getId())
11
        .eventType("ORDER_CREATED")
12
        .payload(JsonSerializer.serialize(order))
13
        .status(0)
14
        .build();
15
    outboxMapper.insert(msg);
16

17
    return order;
18
}
19

20
// 定时任务：扫描未发送消息，投递到 Kafka
21
@Scheduled(fixedDelay = 1000)
22
public void relayOutboxMessages() {
23
    List<OutboxMessage> messages = outboxMapper
24
        .selectPendingMessages(100);
25

26
    for (OutboxMessage msg : messages) {
27
        try {
28
            kafkaTemplate.send("order-events", msg).get();
29
            outboxMapper.updateStatus(msg.getId(), 1);
30
        } catch (Exception e) {
31
            outboxMapper.incrementRetry(msg.getId());
32
            if (msg.getRetryCount() >= 5) {
33
                outboxMapper.updateStatus(msg.getId(), 2);
34
                alertService.send("消息投递失败: " + msg.getId());
35
            }
36
        }
37
    }
38
}

1
{
2
  "mappings": {
3
    "properties": {
4
      "id":            { "type": "long" },
5
      "name": {
6
        "type": "text",
7
        "analyzer": "ik_max_word",
8
        "search_analyzer": "ik_smart",
9
        "fields": {
10
          "keyword": { "type": "keyword" },
11
          "pinyin":  { "type": "text", "analyzer": "pinyin" }
12
        }
13
      },
14
      "category_path": { "type": "keyword" },
15
      "price":         { "type": "scaled_float", "scaling_factor": 100 },
16
      "sales_count":   { "type": "integer" },
17
      "suggest":       { "type": "completion", "analyzer": "ik_max_word" }
18
    }
19
  },
20
  "settings": {
21
    "number_of_shards": 8,
22
    "number_of_replicas": 1
23
  }
24
}

1
# Canal Instance 配置
2
canal.instance.master.address: mysql-master:3306
3
canal.instance.dbUsername: canal
4
canal.instance.filter.regex: "product_db\\..*,order_db\\..*"
5
canal.instance.tsdb.enable: true
6

7
# Kafka Topic 投递
8
canal.mq.servers: kafka:9092
9
canal.mq.topic: canal_${database}_${table}
10
canal.mq.partitionHash: "product_db.products:id,order_db.orders:user_id"

1
#!/bin/bash
2
# 全量备份脚本（Xtrabackup）
3
BACKUP_DIR="/data/backups/mysql/$(date +%Y%m%d)"
4
mkdir -p "$BACKUP_DIR"
5

6
xtrabackup --backup --target-dir="$BACKUP_DIR/full" \
7
  --user=backup --password=$BACKUP_PASS --parallel=4 --compress
8

9
# 上传到对象存储 + 清理 30 天前本地备份
10
rclone sync "$BACKUP_DIR" remote:mysql-backup/$(date +%Y%m%d) --transfers=4
11
find /data/backups/mysql -mtime +30 -exec rm -rf {} +

1
# Prometheus 告警规则
2
groups:
3
  - name: database_alerts
4
    rules:
5
      - alert: MySQLReplicationLag
6
        expr: mysql_slave_seconds_behind_master > 10
7
        for: 2m
8
        labels:
9
          severity: warning
10
        annotations:
11
          summary: "MySQL 复制延迟超过 10 秒"
12

13
      - alert: MySQLConnectionPoolExhausted
14
        expr: mysql_connections_used / mysql_connections_max > 0.9
15
        for: 1m
16
        labels:
17
          severity: critical
18
        annotations:
19
          summary: "MySQL 连接池使用率超过 90%"
20

21
      - alert: RedisMemoryUsage
22
        expr: redis_memory_used_bytes / redis_memory_max_bytes > 0.85
23
        for: 5m
24
        labels:
25
          severity: warning
26
        annotations:
27
          summary: "Redis 内存使用率超过 85%"
28

29
      - alert: OrderCreationLatency
30
        expr: histogram_quantile(0.99, rate(order_create_duration_bucket[5m])) > 0.5
31
        for: 3m
32
        labels:
33
          severity: critical
34
        annotations:
35
          summary: "订单创建 P99 延迟超过 500ms"