1
# ZooKeeper：CP 系统的典型代表
2
# 当 Leader 不可用时，集群进入选举状态，期间拒绝服务
3
# 客户端请求会超时，直到新 Leader 选出
4

5
# etcd 读写流程（简化）
6
etcdctl put /config/key "value"   # 写入需多数节点确认
7
etcdctl get /config/key           # 读取保证线性一致性
8

9
# 如果发生网络分区，少数派分区将无法提供服务
10
# 多数派分区仍可正常读写

1
# Cassandra：AP 系统的典型代表
2
# 写入时指定一致性级别
3
# ANY — 只需一个节点确认（最高可用性，最低一致性）
4
# ONE — 只需一个副本确认
5
# QUORUM — 需要多数副本确认
6
# ALL — 需要所有副本确认（最高一致性，最低可用性）
7

8
# 写入时选择 QUORUM，读取时也选择 QUORUM
9
# 可以保证"读到的值一定比旧值新"（W + R > N）
10
cqlsh> CONSISTENCY QUORUM;
11
cqlsh> INSERT INTO users (id, name) VALUES (1, '张三');
12
cqlsh> SELECT * FROM users WHERE id = 1;

1
-- TiDB：透明的分布式 SQL
2
-- 应用层几乎不需要修改 SQL，TiDB 自动处理分布式查询
3

4
-- 创建表（自动按主键范围分区）
5
CREATE TABLE orders (
6
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
7
    user_id BIGINT NOT NULL,
8
    amount DECIMAL(10,2),
9
    status VARCHAR(20),
10
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
11
    INDEX idx_user (user_id)
12
);
13

14
-- 分布式事务：跨分区操作自动使用 2PC
15
BEGIN;
16
UPDATE accounts SET balance = balance - 100 WHERE id = 1;
17
UPDATE accounts SET balance = balance + 100 WHERE id = 2;
18
COMMIT;
19
-- TiDB 保证分布式 ACID，应用层无需感知分片

1
# 典型的读写分离模式
2
class ProductService:
3
    def get_product(self, product_id: str):
4
        # 1. 先查缓存
5
        product = redis.get(f"product:{product_id}")
6
        if product:
7
            return json.loads(product)
8

9
        # 2. 缓存未命中，查主库
10
        product = mysql.query(
11
            "SELECT * FROM products WHERE id = %s", product_id
12
        )
13
        if product:
14
            # 3. 回填缓存（设置 TTL 防止数据过期）
15
            redis.setex(
16
                f"product:{product_id}",
17
                300,  # 5 分钟 TTL
18
                json.dumps(product)
19
            )
20
        return product
21

22
    def search_products(self, keyword: str):
23
        # 搜索走 Elasticsearch，不走主库
24
        return es.search(
25
            index="products",
26
            body={"query": {"match": {"name": keyword}}}
27
        )

1
# 双写：应用层同时写入 MySQL 和 Elasticsearch
2
def create_order(order):
3
    # 问题：如果 ES 写入失败，MySQL 已提交，数据不一致
4
    mysql.insert("orders", order)          # 成功
5
    es.index("orders", order)              # 失败 → 数据不一致
6

7
# 改进：使用本地消息表保证最终一致性
8
def create_order_safe(order):
9
    # 1. 写入主库 + 本地消息表（同一事务）
10
    with mysql.transaction():
11
        mysql.insert("orders", order)
12
        mysql.insert("outbox", {
13
            "aggregate_id": order.id,
14
            "event_type": "order_created",
15
            "payload": json.dumps(order),
16
            "status": "pending"
17
        })
18
    # 2. 异步消费者读取 outbox 表，写入 ES
19
    # 3. 写入成功后标记 outbox 为 completed

1
# Debezium + Kafka CDC 配置示例
2
# Debezium 捕获 MySQL Binlog，发送到 Kafka，再写入 ES
3

4
version: "3.8"
5
services:
6
  # MySQL 源库
7
  mysql:
8
    image: mysql:8.0
9
    environment:
10
      MYSQL_ROOT_PASSWORD: root
11
    command: --log-bin=mysql-bin --binlog-format=ROW
12
    ports:
13
      - "3306:3306"
14

15
  # Debezium Connector
16
  connect:
17
    image: debezium/connect:2.5
18
    environment:
19
      BOOTSTRAP_SERVERS: kafka:9092
20
      GROUP_ID: connect-cluster
21
    ports:
22
      - "8083:8083"
23

24
  # 注册 MySQL Connector
25
  # curl -X POST http://localhost:8083/connectors \
26
  #   -H "Content-Type: application/json" \
27
  #   -d '{
28
  #     "name": "mysql-connector",
29
  #     "config": {
30
  #       "connector.class": "io.debezium.connector.mysql.MySqlConnector",
31
  #       "database.hostname": "mysql",
32
  #       "database.port": "3306",
33
  #       "database.user": "root",
34
  #       "database.password": "root",
35
  #       "database.server.id": "1",
36
  #       "database.server.name": "mysql_prod",
37
  #       "database.include.list": "ecommerce",
38
  #       "table.include.list": "ecommerce.orders,ecommerce.products"
39
  #     }
40
  #   }'

1
# POC 验证脚本框架
2
#!/bin/bash
3
# 数据库选型 POC 自动化测试
4

5
# 1. 基准性能测试
6
echo "=== 1. 性能基准 ==="
7
sysbench oltp_read_write \
8
  --db-driver=mysql \
9
  --mysql-host=localhost \
10
  --threads=32 \
11
  --tables=10 \
12
  --table-size=1000000 \
13
  run
14

15
# 2. 故障恢复测试
16
echo "=== 2. 故障恢复 ==="
17
# 模拟节点宕机，验证 RTO
18
docker stop db-node-2
19
sleep 30  # 等待故障检测
20
docker start db-node-2
21
# 测量恢复时间
22

23
# 3. 数据一致性验证
24
echo "=== 3. 一致性测试 ==="
25
# 并发写入后，验证所有副本数据一致
26
for i in $(seq 1 1000); do
27
  mysql -e "INSERT INTO test VALUES ($i, 'data_$i')"
28
done
29
# 比对主从数据

1
# 电商核心：库存扣减的强一致性保证
2
def deduct_stock(product_id: str, quantity: int) -> bool:
3
    """库存扣减必须在同一事务中完成"""
4
    with mysql.transaction():
5
        # 悲观锁：SELECT ... FOR UPDATE 防止超卖
6
        stock = mysql.query(
7
            "SELECT stock FROM products WHERE id = %s FOR UPDATE",
8
            product_id
9
        )
10
        if stock < quantity:
11
            return False  # 库存不足
12
        mysql.execute(
13
            "UPDATE products SET stock = stock - %s WHERE id = %s",
14
            quantity, product_id
15
        )
16
        # 创建订单记录（同一事务）
17
        mysql.execute(
18
            "INSERT INTO orders (product_id, quantity, status) "
19
            "VALUES (%s, %s, 'created')",
20
            product_id, quantity
21
        )
22
    return True

1
# 社交系统：Feed 流的写入与读取
2
def publish_post(user_id: str, content: str):
3
    """发布动态：写入作者时间线 + 粉丝时间线（扇出写模式）"""
4
    post_id = generate_id()
5
    # 1. 写入动态表
6
    cassandra.insert("posts", {
7
        "id": post_id, "user_id": user_id,
8
        "content": content, "created_at": now()
9
    })
10
    # 2. 写入作者自己的时间线
11
    cassandra.insert("timeline", {
12
        "user_id": user_id, "post_id": post_id,
13
        "created_at": now()
14
    })
15
    # 3. 扇出写入粉丝时间线（异步）
16
    followers = redis.smembers(f"followers:{user_id}")
17
    for follower_id in followers:
18
        cassandra.insert_async("timeline", {
19
            "user_id": follower_id, "post_id": post_id,
20
            "created_at": now()
21
        })

1
-- IoT 系统：InfluxDB 时序数据查询
2
-- 查询最近 1 小时每 5 分钟的平均温度
3
SELECT mean("temperature") AS "avg_temp"
4
FROM "sensor_data"
5
WHERE time > now() - 1h
6
  AND "device_id" = 'sensor-001'
7
GROUP BY time(5m)
8

9
-- 连续查询：自动降采样（保留原始数据 30 天，聚合数据永久保留）
10
CREATE CONTINUOUS QUERY "cq_5m_avg" ON "iot_db"
11
BEGIN
12
  SELECT mean("temperature") AS "avg_temp",
13
         mean("humidity") AS "avg_humidity"
14
  INTO "downsampled"."sensor_5m"
15
  FROM "sensor_data"
16
  GROUP BY time(5m), "device_id"
17
END

1
# 日志系统：ELK Stack 典型配置
2
# Logstash → Elasticsearch → Kibana
3

4
# Logstash 配置：解析 Nginx 访问日志
5
input:
6
  file:
7
    path: "/var/log/nginx/access.log"
8
    start_position: "beginning"
9

10
filter:
11
  grok:
12
    match: { "message": "%{COMBINEDAPACHELOG}" }
13
  date:
14
    match: [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
15

16
output:
17
  elasticsearch:
18
    hosts: ["http://elasticsearch:9200"]
19
    index: "nginx-access-%{+YYYY.MM.dd}"
20
  # 同时写入 Redis 缓存最近日志
21
  redis:
22
    host: "redis"
23
    data_type: "list"
24
    key: "logstash:recent"