数据存储与中间件层

          +-----------+
          |  应用层    |
          +-----+-----+
                |
          +-----+-----+
          |  读写分离   |
          |  中间件     |  (ShardingSphere/MyCat/ProxySQL)
          +--+------+--+
             |      |
     +-------+--+  ++--------+
     |  Master  |  |  Slave   |
     | (写)     |  | (读)     |
     +----------+  +----------+
                    |  Slave   |
                    | (读)     |
                    +----------+

-- 按用户ID取模分库: user_db_0, user_db_1, user_db_2
-- 分库键: user_id % 3

-- 按时间分表: order_202601, order_202602
-- 分表键: create_time

-- 按地域分库: db_asia, db_europe, db_america
-- 分库键: region_code

# sharding-sphere-config.yaml
dataSources:
  ds_0:
    url: jdbc:mysql://localhost:3306/db_0?useSSL=false
    username: root
    password: ${MYSQL_PASSWORD}
  ds_1:
    url: jdbc:mysql://localhost:3306/db_1?useSSL=false
    username: root
    password: ${MYSQL_PASSWORD}

rules:
  - !SHARDING
    tables:
      t_order:
        actualDataNodes: ds_${0..1}.t_order_${0..1}
        databaseStrategy:
          standard:
            shardingColumn: user_id
            shardingAlgorithmName: db_inline_mod
        tableStrategy:
          standard:
            shardingColumn: order_id
            shardingAlgorithmName: table_inline_mod
    shardingAlgorithms:
      db_inline_mod:
        type: INLINE
        props:
          algorithm-expression: ds_${user_id % 2}
      table_inline_mod:
        type: INLINE
        props:
          algorithm-expression: t_order_${order_id % 2}

聚簇索引 (Primary Key)
  [10|...] → [20|...] → [30|...] → ...
    ↓           ↓           ↓
  数据行      数据行      数据行

辅助索引 (Secondary Index)
  [name='Alice'|pk=10] → [name='Bob'|pk=20] → ...
          ↓                       ↓
      回表查询聚簇索引          回表查询聚簇索引

事务100: INSERT
  → 事务200: UPDATE name='Bob'
    → 事务300: UPDATE name='Charlie'
      → (最新版本)

ReadView结构:
  - m_ids:   活跃事务ID列表（启动时未提交的）
  - min_trx_id: 最小活跃事务ID
  - max_trx_id: 预分配的下一个事务ID
  - creator_trx_id: 创建该ReadView的事务ID

判断规则:
  1. DB_TRX_ID == creator_trx_id → 可见（自己修改的）
  2. DB_TRX_ID < min_trx_id  → 可见（已提交的）
  3. DB_TRX_ID >= max_trx_id → 不可见（未来事务）
  4. DB_TRX_ID ∈ m_ids      → 不可见（未提交）

-- 创建联合索引
CREATE INDEX idx_org_status_time ON t_order (org_id, status, create_time);

-- 走索引的情况（最左前缀匹配）
WHERE org_id = 100;                                 -- 只使用org_id
WHERE org_id = 100 AND status = 1;                  -- 使用org_id + status
WHERE org_id = 100 AND status = 1 AND create_time > '2026-01-01'; -- 完全使用

-- 不走索引或部分走索引的情况
WHERE status = 1;                                   -- 无最左列，全表扫描
WHERE org_id = 100 AND create_time > '2026-01-01';  -- 只用到org_id，跳过status

-- 表结构
CREATE TABLE t_user (
  id BIGINT PRIMARY KEY,
  name VARCHAR(50),
  age INT,
  email VARCHAR(100),
  INDEX idx_name_age (name, age)
);

-- 覆盖索引查询：SQL所需字段完全在索引中，无需回表
EXPLAIN SELECT name, age FROM t_user WHERE name = 'Alice';
-- Extra: Using index  ← 表示使用了覆盖索引

-- 非覆盖索引查询：需要回表
EXPLAIN SELECT name, email FROM t_user WHERE name = 'Alice';
-- Extra: Using index condition  ← 需要回表查询email

-- MySQL 5.6+ 特性：将WHERE条件从Server层"下推"到存储引擎层过滤

-- 示例：联合索引 idx_age_name (age, name)
SELECT * FROM t_user WHERE age > 20 AND name LIKE '%Li%';

-- 无ICP时:
--   存储引擎扫描age>20的记录 → 回表 → Server层过滤name LIKE
-- 有ICP时:
--   存储引擎同时用age和name条件过滤 → 减少回表次数

-- 优化前
EXPLAIN SELECT * FROM t_order WHERE status = 0 ORDER BY create_time DESC;
-- type: ALL, Extra: Using filesort  ← 全表扫描+额外排序，危险！

-- 优化后：创建联合索引
CREATE INDEX idx_status_create_time ON t_order (status, create_time);
EXPLAIN SELECT * FROM t_order WHERE status = 0 ORDER BY create_time DESC;
-- type: ref, Extra: NULL  ← 索引定位+索引有序，高效！

-- =============================================
-- 企业SQL编写规范示例
-- =============================================

-- 1. 禁止使用SELECT *，必须指定字段列表
-- 错误: SELECT * FROM t_user;
-- 正确:
SELECT id, name, age FROM t_user;

-- 2. 大表查询必须带LIMIT
SELECT id, name FROM t_order WHERE status = 1 LIMIT 100;

-- 3. JOIN不宜超过3张表，必须确保关联字段有索引
SELECT o.id, u.name
FROM t_order o
INNER JOIN t_user u ON o.user_id = u.id   -- user_id和id都有索引
WHERE o.create_time > '2026-01-01';

-- 4. WHERE条件中禁止对索引列做函数操作
-- 错误: WHERE DATE(create_time) = '2026-01-01'
-- 正确: WHERE create_time >= '2026-01-01' AND create_time < '2026-01-02'

-- 5. 分页查询深翻页优化
-- 错误（深翻页）: SELECT * FROM t_order LIMIT 1000000, 20;
-- 正确（子查询优化）:
SELECT * FROM t_order
WHERE id > (SELECT id FROM t_order ORDER BY id LIMIT 1000000, 1)
ORDER BY id LIMIT 20;

-- 6. UNION代替OR（OR导致索引失效）
-- 错误: SELECT * FROM t_user WHERE name = 'Alice' OR age = 25;
-- 正确:
SELECT * FROM t_user WHERE name = 'Alice'
UNION
SELECT * FROM t_user WHERE age = 25;

-- 7. 批量INSERT使用一条语句
INSERT INTO t_order (id, user_id, amount) VALUES
(1, 101, 99.9),
(2, 102, 199.9),
(3, 103, 299.9);

┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐
│  发现    │ →  │  分析    │ →  │  优化    │ →  │  验证    │
│          │    │          │    │          │    │          │
│ slow_log  │    │ EXPLAIN  │    │ 加索引    │    │ EXPLAIN  │
│ 慢查询日志 │    │ profile  │    │ 改写SQL   │    │ 响应时间  │
│ 告警系统   │    │ 表结构   │    │ 分表     │    │ QPS监控   │
└──────────┘    └──────────┘    └──────────┘    └──────────┘

-- 查看慢查询配置
SHOW VARIABLES LIKE 'slow_query%';
SHOW VARIABLES LIKE 'long_query_time';

-- 配置（my.cnf）
slow_query_log = ON
slow_query_log_file = /var/log/mysql/slow.log
long_query_time = 1        -- 超过1秒的记录
log_queries_not_using_indexes = ON

# 分析慢查询日志
pt-query-digest /var/log/mysql/slow.log > slow_analysis.txt

# 关键指标：Query_time、Lock_time、Rows_examined、Rows_sent

# application.yml
spring:
  datasource:
    hikari:
      maximum-pool-size: 20          # 最大连接数（核心配置）
      minimum-idle: 10               # 最小空闲连接
      idle-timeout: 300000           # 空闲超时（5分钟）
      connection-timeout: 30000      # 连接超时（30秒）
      max-lifetime: 1800000          # 最大存活时间（30分钟）
      pool-name: HikariPool-Main     # 连接池名称

spring:
  datasource:
    druid:
      initial-size: 5               # 初始化连接数
      min-idle: 5                   # 最小空闲连接
      max-active: 20                # 最大活跃连接
      max-wait: 60000               # 获取连接等待超时（毫秒）
      time-between-eviction-runs-millis: 60000   # 检查间隔
      validation-query: SELECT 1    # 心跳SQL
      test-while-idle: true         # 空闲时验证
      test-on-borrow: false         # 获取时验证（影响性能）
      test-on-return: false         # 归还时验证
      filter:
        stat:
          enabled: true
          log-slow-sql: true
          slow-sql-millis: 1000

// Spring Data Redis 示例
@Service
public class UserCacheService {
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    // 缓存穿透保护：缓存空值
    public User getUserById(Long id) {
        String key = "user:" + id;

        // 1. 查缓存
        User user = (User) redisTemplate.opsForValue().get(key);
        if (user != null) {
            return user;
        }

        // 2. 缓存空值保护（缓存穿透）
        String nullFlag = (String) redisTemplate.opsForValue().get(key + ":null");
        if (nullFlag != null) {
            return null;  // 防止穿透到DB
        }

        // 3. 查数据库
        user = userMapper.selectById(id);
        if (user == null) {
            // 设置空值缓存，TTL 60秒
            redisTemplate.opsForValue().set(key + ":null", "1", 60, TimeUnit.SECONDS);
        } else {
            // 正常缓存，TTL 30分钟
            redisTemplate.opsForValue().set(key, user, 30, TimeUnit.MINUTES);
        }
        return user;
    }
}

// Hash适合存储结构化对象，比String + JSON更节省内存，支持字段单独操作
public void updateUserField(Long userId, String field, Object value) {
    String key = "user:" + userId;
    redisTemplate.opsForHash().put(key, field, value);
}

// 生产者
public void sendTask(Long taskId) {
    redisTemplate.opsForList().rightPush("task:queue", taskId.toString());
}

// 消费者（阻塞读取）
public void consumeTask() {
    while (true) {
        String taskId = (String) redisTemplate.opsForList()
            .leftPop("task:queue", 5, TimeUnit.SECONDS);
        if (taskId != null) {
            processTask(Long.parseLong(taskId));
        }
    }
}

// 积分排行榜
public void addScore(Long userId, double score) {
    redisTemplate.opsForZSet().add("ranking:score", userId.toString(), score);
}

// 获取TOP 10
public Set<String> getTop10() {
    return redisTemplate.opsForZSet()
        .reverseRange("ranking:score", 0, 9);
}

// 获取用户排名
public Long getUserRank(Long userId) {
    return redisTemplate.opsForZSet()
        .reverseRank("ranking:score", userId.toString());
}

// Stream相比List的优势：支持消费者组、消息持久化、ACK机制
// 适合需要可靠消费的场景
redisTemplate.opsForStream()
    .add(StreamRecords.newRecord()
        .ofObject(message)
        .withStreamKey("stream:orders"));

// 布隆过滤器防止缓存穿透
@Component
public class BloomFilterService {
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    private static final String BLOOM_KEY = "bloom:users";
    private static final int EXPECTED_SIZE = 100_0000;
    private static final double FPP = 0.01;

    private BloomFilter<Long> bloomFilter;

    @PostConstruct
    public void init() {
        bloomFilter = BloomFilter.create(
            Funnels.longFunnel(), EXPECTED_SIZE, FPP);
        // 从DB加载已有用户ID到布隆过滤器
        List<Long> allUserIds = userMapper.selectAllIds();
        allUserIds.forEach(bloomFilter::put);
    }

    public boolean mightContain(Long userId) {
        return bloomFilter.mightContain(userId);
    }
}

// 互斥锁防止缓存击穿
public User queryWithMutex(Long id) {
    String key = "user:" + id;
    // 1. 查缓存
    User user = (User) redisTemplate.opsForValue().get(key);
    if (user != null) return user;

    // 2. 获取互斥锁
    String lockKey = "lock:user:" + id;
    Boolean locked = redisTemplate.opsForValue()
        .setIfAbsent(lockKey, "1", 10, TimeUnit.SECONDS);

    if (Boolean.TRUE.equals(locked)) {
        try {
            // 3. 查DB
            user = userMapper.selectById(id);
            redisTemplate.opsForValue().set(key, user, 30, TimeUnit.MINUTES);
            return user;
        } finally {
            redisTemplate.delete(lockKey);
        }
    } else {
        // 4. 没拿到锁，休眠后重试
        Thread.sleep(100);
        return queryWithMutex(id);
    }
}

// TTL加随机值防止雪崩
int baseTTL = 30 * 60;  // 30分钟
int randomTTL = baseTTL + new Random().nextInt(10 * 60);  // 增加0~10分钟随机值
redisTemplate.opsForValue().set(key, user, randomTTL, TimeUnit.SECONDS);

@Configuration
public class RedissonConfig {
    @Bean
    public RedissonClient redissonClient() {
        Config config = new Config();
        config.useSingleServer()
            .setAddress("redis://localhost:6379")
            .setConnectionPoolSize(10);
        return Redisson.create(config);
    }
}

@Service
public class OrderService {
    @Autowired
    private RedissonClient redissonClient;

    // 标准分布式锁
    public void createOrder(Long orderId) {
        RLock lock = redissonClient.getLock("order:" + orderId);
        try {
            // 尝试加锁，最多等3秒，锁有效30秒
            if (lock.tryLock(3, 30, TimeUnit.SECONDS)) {
                // 执行业务逻辑
                processOrder(orderId);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            lock.unlock();
        }
    }

    // 看门狗机制（自动续期）
    public void processWithWatchdog(Long orderId) {
        RLock lock = redissonClient.getLock("order:" + orderId);
        lock.lock(); // 默认30秒，看门狗每10秒续期一次
        try {
            // 执行业务逻辑（即使超过30秒也不用担心锁超时）
            longProcess(orderId);
        } finally {
            lock.unlock();
        }
    }

    // 可重入锁
    public void reentrantExample(Long userId) {
        RLock lock = redissonClient.getLock("user:" + userId);
        lock.lock();
        try {
            // 第一层
            innerMethod(userId);  // 内部再次加同一把锁，可重入
        } finally {
            lock.unlock();
        }
    }

    private void innerMethod(Long userId) {
        RLock lock = redissonClient.getLock("user:" + userId);
        lock.lock();  // 可重入：同一个线程可多次加锁
        try {
            // 内部逻辑
        } finally {
            lock.unlock();
        }
    }

    // 红锁（多节点分布式锁，RedLock算法）
    public void redLockExample(Long resourceId) {
        RLock lock1 = redissonClient.getLock("node1:resource:" + resourceId);
        RLock lock2 = redissonClient.getLock("node2:resource:" + resourceId);
        RLock lock3 = redissonClient.getLock("node3:resource:" + resourceId);
        RLock redLock = new RedissonRedLock(lock1, lock2, lock3);

        try {
            if (redLock.tryLock(5, 30, TimeUnit.SECONDS)) {
                // 获得多数节点锁，执行业务
                processResource(resourceId);
            }
        } finally {
            redLock.unlock();
        }
    }
}

# 使用redis-cli --bigkeys 扫描
redis-cli --bigkeys -h 127.0.0.1 -p 6379

# 手动查看Key大小
MEMORY USAGE user:123456
STRLEN big_string_key
HLEN big_hash_key

// 1. 本地缓存 + Redis两级缓存（Caffeine + Redis）
@Configuration
public class TwoLevelCacheConfig {
    @Bean
    public Cache<String, Object> caffeineCache() {
        return Caffeine.newBuilder()
            .maximumSize(10000)
            .expireAfterWrite(5, TimeUnit.MINUTES)
            .build();
    }
}

@Service
public class HotKeyService {
    @Autowired
    private Cache<String, Object> caffeineCache;
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    public Object getHotData(String key) {
        // 1. 查本地缓存
        Object value = caffeineCache.getIfPresent(key);
        if (value != null) return value;

        // 2. 查Redis
        value = redisTemplate.opsForValue().get(key);
        if (value != null) {
            caffeineCache.put(key, value);
        }
        return value;
    }
}

// 2. 热Key散列：将单个Key拆分为多个
// 原Key: hot_product:12345
// 拆分后: hot_product:12345:0, hot_product:12345:1, ... hot_product:12345:9
// 读写时随机选择一个子Key

中小企业、快速集成    → RabbitMQ（简单、生态好）
金融/订单/事务场景     → RocketMQ（事务消息、可靠）
大数据/日志/流处理     → Kafka（高吞吐、海量数据）
IoT/物联网            → RabbitMQ（MQTT协议支持）

          +-----------+
          |  Producer  |
          +-----+-----+
                |
          +-----+-----+
          |  Exchange  |
          +--+------+--+
             |      |
    +--------+-+  +-+--------+
    | Queue A  |  | Queue B  |
    +--------+-+  +-+--------+
             |      |
    +--------+-+  +-+--------+
    |Consumer A|  |Consumer B|
    +----------+  +----------+

# application.yml
spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: /
    # 生产者确认
    publisher-confirm-type: correlated   # 异步回调确认
    publisher-returns: true              # 消息不可达返回
    # 消费者
    listener:
      simple:
        acknowledge-mode: manual          # 手动ACK
        prefetch: 1                      # 每次拉取1条
        retry:
          enabled: true
          max-attempts: 3

@Configuration
public class RabbitMQConfig {
    // 延迟队列：使用死信队列实现
    @Bean
    public Queue delayQueue() {
        return QueueBuilder.durable("delay.queue")
            .ttl(30000)                        // 30秒过期
            .deadLetterExchange("direct.exchange")
            .deadLetterRoutingKey("process")
            .build();
    }

    @Bean
    public Queue processQueue() {
        return new Queue("process.queue");
    }
}

@Service
public class OrderMessageService {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    // 发送订单消息（带确认回调）
    public void sendOrderMessage(Order order) {
        CorrelationData correlationData = new CorrelationData(order.getOrderId());
        rabbitTemplate.convertAndSend(
            "order.exchange", "order.create", order, correlationData);

        // 异步确认回调
        rabbitTemplate.setConfirmCallback((correlation, ack, cause) -> {
            if (!ack) {
                log.error("消息发送失败: {}", cause);
                // 补偿：重发或记录到数据库
            }
        });
    }

    // 消费订单消息（手动ACK）
    @RabbitListener(queues = "process.queue")
    public void processOrder(Order order, Channel channel,
                            @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        try {
            // 业务处理
            orderService.process(order);
            // 手动ACK
            channel.basicAck(tag, false);
        } catch (Exception e) {
            // 拒绝消息，不重新入队（进入死信队列）
            channel.basicNack(tag, false, false);
        }
    }
}

// 常见延迟场景：订单30分钟未支付自动取消
// 方案：RabbitMQ死信队列（TTL过期 → 死信队列 → 消费者）
//
// 生产流程：
// 1. 下单 → 发送消息到 delay.queue（TTL=30分钟）
// 2. 30分钟后消息过期 → 转发到 dead.exchange
// 3. dead.exchange 路由到 cancel.queue
// 4. 取消订单消费者处理

// 也可以使用 rabbitmq-delayed-message-exchange 插件
@Bean
public CustomExchange delayExchange() {
    Map<String, Object> args = new HashMap<>();
    args.put("x-delayed-type", "direct");
    return new CustomExchange("delay.exchange", "x-delayed-message",
        true, false, args);
}

事务消息流程:

Producer                  Broker
    |                        |
    |-- 1. 发送半消息 -------→|  (Half Message，暂不可见)
    |←--- 2. OK -------------|
    |                        |
    |-- 3. 执行本地事务 ------|
    |                        |
    |-- 4. COMMIT/ROLLBACK --→|  (提交或回滚)
    |                        |
    |    5. 回查(最多15次)    |
    |←--- 回查接口 -----------|
    |-- 回查结果 ------------→|
    |                        |
    |         6. 消息可见     |
    |                        → Consumer

// RocketMQ事务消息示例
@Component
public class OrderTransactionListener implements TransactionListener {
    @Autowired
    private OrderService orderService;

    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        try {
            // 执行本地事务（创建订单）
            orderService.createOrder((Order) arg);
            return LocalTransactionState.COMMIT_MESSAGE;
        } catch (Exception e) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // 回查：检查本地事务是否成功
        String orderId = msg.getKeys();
        boolean success = orderService.checkOrderExists(orderId);
        return success ?
            LocalTransactionState.COMMIT_MESSAGE :
            LocalTransactionState.ROLLBACK_MESSAGE;
    }
}

// 顺序消息：同一个订单的消息必须依次消费
// 原理：相同MessageQueue的消息串行消费
// 策略：使用订单ID作为Sharding Key

// 生产者
producer.send(message, (mqs, msg, arg) -> {
    String orderId = (String) arg;
    int index = Math.abs(orderId.hashCode()) % mqs.size();
    return mqs.get(index);  // 相同orderId进入同一个Queue
}, orderId);

// RocketMQ原生支持18个级别的延时
// 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

Message msg = new Message("order_topic", "cancel", orderJson.getBytes());
msg.setDelayTimeLevel(5);  // 5 = 1分钟（取消订单延迟1分钟）
producer.send(msg);

# Kafka核心概念
# Topic → Partition (有序、可并行) → Replica (副本)
# Consumer Group (组内负载均衡消费)

# 日志收集架构
# App → Logback/KafkaAppender → Kafka → Logstash → ES → Kibana

# CDC (Change Data Capture)
# MySQL Binlog → Canal → Kafka → 下游服务/ES/数仓

spring:
  kafka:
    bootstrap-servers: localhost:9092,localhost:9093
    consumer:
      group-id: order-group
      auto-offset-reset: earliest    # 从头消费
      enable-auto-commit: false      # 手动提交
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      acks: all                      # 最强可靠性
      retries: 3

┌─────────┐   Confirm    ┌─────────┐  持久化    ┌─────────┐
│Producer │ ───────────→ │ Broker  │ ────────→ │Consumer │
│         │              │         │           │         │
│ 重试机制 │              │ 同步刷盘 │           │ 手动ACK │
│ 事务消息 │              │ 主从复制 │           │ 幂等消费 │
│ 回调确认 │              │ OFFSET   │           │ 重试机制 │
└─────────┘              └─────────┘           └─────────┘

# RocketMQ控制台或CLI
mqadmin consumerProgress -n 127.0.0.1:9876 -g order_consumer_group

# Kafka Consumer Group状态
kafka-consumer-groups --bootstrap-server localhost:9092 \
  --group order-group --describe
# 关注 LAG（未消费消息数）

// 幂等消费：业务层使用唯一键去重
@Component
public class IdempotentConsumer {
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    public boolean consume(String messageId, Consumer<Message> consumer) {
        String key = "msg:idempotent:" + messageId;
        // SETNX原子操作，保证只执行一次
        Boolean success = redisTemplate.opsForValue()
            .setIfAbsent(key, "1", 1, TimeUnit.DAYS);

        if (Boolean.TRUE.equals(success)) {
            consumer.accept(message);
            return true;
        }
        log.info("消息已消费，跳过: {}", messageId);
        return false;
    }
}

// 方案一：单一队列/分区（性能受限）
// RabbitMQ：所有消息发到同一个Queue
// Kafka：所有消息使用同一Partition Key
// RocketMQ：相同订单ID使用同一MessageQueue

// 方案二：业务层排序（推荐）
// 将顺序要求拆解到业务层面：
// 1. 为每个消息携带 sequenceId
// 2. 消费者按sequenceId排序处理
// 3. 乱序的消息暂存到缓冲区

// Product索引Mapping示例
PUT /products
{
  "settings": {
    "number_of_shards": 3,
    "number_of_replicas": 1,
    "refresh_interval": "10s",
    "analysis": {
      "analyzer": {
        "ik_smart_analyzer": {
          "type": "custom",
          "tokenizer": "ik_smart"
        }
      }
    }
  },
  "mappings": {
    "properties": {
      "productId": {
        "type": "keyword"           // 精确匹配
      },
      "productName": {
        "type": "text",             // 全文检索
        "analyzer": "ik_smart_analyzer",
        "fields": {
          "keyword": {
            "type": "keyword"       // 子字段：精确匹配
          }
        }
      },
      "categoryId": {
        "type": "long"
      },
      "price": {
        "type": "double"
      },
      "stock": {
        "type": "integer"
      },
      "status": {
        "type": "byte"
      },
      "createTime": {
        "type": "date",
        "format": "yyyy-MM-dd HH:mm:ss"
      },
      "tags": {
        "type": "keyword"           // 数组字段
      },
      "description": {
        "type": "text",
        "analyzer": "ik_smart_analyzer"
      }
    }
  }
}

MySQL Binlog → Canal → MQ (Kafka/RocketMQ) → 消费程序 → ES
                                                     → 缓存 (可选)
                                                     → 其他下游

# canal.properties
canal.serverMode = tcp
canal.destinations = example

# instance.properties
canal.instance.master.address = 127.0.0.1:3306
canal.instance.dbUsername = canal
canal.instance.dbPassword = canal
canal.instance.connectionCharset = UTF-8
canal.instance.filter.regex = mydb\\.t_product
canal.mq.topic = es_product_topic

@Component
public class ProductSyncConsumer {
    @Autowired
    private ElasticsearchRestTemplate esTemplate;

    @KafkaListener(topics = "es_product_topic", groupId = "es-sync-group")
    public void syncProduct(String message) {
        CanalMessage canalMsg = JSON.parseObject(message, CanalMessage.class);

        switch (canalMsg.getType()) {
            case "INSERT":
            case "UPDATE":
                Product product = JSON.parseObject(
                    canalMsg.getData().get(0).toString(), Product.class);
                esTemplate.save(product);
                break;
            case "DELETE":
                String productId = canalMsg.getData().get(0)
                    .getString("productId");
                esTemplate.delete(productId, Product.class);
                break;
        }
    }
}

GET /products/_search
{
  "size": 20,
  "query": {
    "bool": {
      "must": [
        {
          "match": {
            "productName": "手机 华为"      // 全文匹配
          }
        }
      ],
      "filter": [
        {
          "term": {
            "status": 1                    // 精确过滤
          }
        },
        {
          "range": {
            "price": {
              "gte": 1000,
              "lte": 8000                // 范围过滤
            }
          }
        }
      ]
    }
  },
  "sort": [
    { "price": { "order": "asc" } }      // 排序
  ],
  "highlight": {
    "fields": {
      "productName": {
        "pre_tags": ["<em>"],
        "post_tags": ["</em>"]
      }
    }
  },
  "aggs": {
    "by_category": {
      "terms": {
        "field": "categoryId",            // 按类目聚合
        "size": 10
      },
      "aggs": {
        "avg_price": {
          "avg": {
            "field": "price"             // 每个类目的平均价格
          }
        }
      }
    }
  }
}

@Service
public class ProductSearchService {
    @Autowired
    private ElasticsearchRestTemplate esTemplate;

    public SearchResponse<Product> searchProducts(
            String keyword, Long categoryId, Double priceMin,
            Double priceMax, int page, int size) {

        NativeSearchQueryBuilder queryBuilder = new NativeSearchQueryBuilder();
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();

        // 全文检索
        if (StringUtils.hasText(keyword)) {
            boolQuery.must(QueryBuilders.matchQuery("productName", keyword));
        }

        // 类目过滤
        if (categoryId != null) {
            boolQuery.filter(QueryBuilders.termQuery("categoryId", categoryId));
        }

        // 价格范围过滤
        if (priceMin != null || priceMax != null) {
            RangeQueryBuilder range = QueryBuilders.rangeQuery("price");
            if (priceMin != null) range.gte(priceMin);
            if (priceMax != null) range.lte(priceMax);
            boolQuery.filter(range);
        }

        // 分页（from + size，避免深分页用search_after）
        queryBuilder.withQuery(boolQuery)
            .withPageable(PageRequest.of(page, size))
            .withSort(Sort.by("price").ascending());

        SearchHits<Product> hits = esTemplate.search(
            queryBuilder.build(), Product.class);
        return new SearchResponse<>(hits);
    }
}

# elasticsearch.yml 配置要点

# 节点角色（7.x+ 角色分离）
node.roles: [ data, master, ingest ]

# 生产环境推荐 3个节点集群：
# Node 1: master + data + ingest
# Node 2: master + data + ingest
# Node 3: master + data + ingest
# 至少 3 个 master-eligible 节点（防止脑裂）

# JVM内存配置
# ES_JAVA_HOME=/usr/share/elasticsearch/jdk
# ES_JAVA_OPTS="-Xms16g -Xmx16g"   # 不超过物理内存50%，不超过32G

// 写入优化
PUT /_template/log_template
{
  "index_patterns": ["logs-*"],
  "settings": {
    "refresh_interval": "30s",          // 降低刷新频率
    "translog.durability": "async",     // 异步持久化
    "translog.sync_interval": "5s",
    "number_of_replicas": 0,            // 写入时副本数设0，写入后改为1
    "routing.allocation.total_shards_per_node": 3
  }
}

// 查询优化
// - 使用filter context代替query context（利用缓存）
// - 减少返回字段（_source选择性返回）
// - 避免script查询
// - 合理设置字段的index_options

// 评论系统：文档结构天然适合
@Document(collection = "comments")
public class Comment {
    @Id
    private String id;
    private Long userId;
    private String content;
    private List<String> images;
    private Date createTime;
    private Map<String, Integer> likes;  // {userId: 1}
}

# Nginx核心配置示例

upstream backend_servers {
    # 负载均衡策略
    # 默认: 轮询
    # ip_hash: 根据IP哈希（Session保持）
    # least_conn: 最小连接数
    # weight: 权重
    server 127.0.0.1:8080 weight=3;
    server 127.0.0.1:8081 weight=2;
    server 127.0.0.1:8082 weight=1 backup;  # 备用
}

server {
    listen 80;
    server_name api.example.com;

    # 动静分离
    location /api/ {
        proxy_pass http://backend_servers;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    }

    location /static/ {
        root /var/www/site;
        expires 30d;
        add_header Cache-Control "public, immutable";
    }

    # 限流
    location /api/order/ {
        limit_req zone=order_limit burst=20 nodelay;
        proxy_pass http://backend_servers;
    }
}

# 限流区域定义
limit_req_zone $binary_remote_addr zone=order_limit:10m rate=100r/s;

# MinIO Docker Compose
services:
  minio:
    image: minio/minio:latest
    ports:
      - "9000:9000"    # API端口
      - "9001:9001"    # Web控制台端口
    environment:
      MINIO_ROOT_USER: admin
      MINIO_ROOT_PASSWORD: ${MINIO_PASSWORD}
    command: server /data --console-address ":9001"
    volumes:
      - ./data:/data