webman 数据库实战：连接池、ORM 选型、事务边界与大数据处理
webman 数据库实战：连接池、ORM 选型、事务边界与大数据处理数据库是 webman 与传统 FPM 框架差异最大的环节。FPM 下每个请求开新连接、请求结束关闭，天然隔离；webman 的每个 worker 进程持有自己的连接并跨请求复用——连接管理从"框架的事"变成了"你的事"。这篇讲透连接池配置、think-orm 与 Eloquent 的选型、事务的正确边界，以及千万级数据的处理姿势。一、常驻内存下的连接模型先看清楚 webman 的数据库连接结构：worker-1 ─── 连接A（MySQL）──┐ worker-2 ─── 连接B ├── MySQL max_connections 上限（如 1000） worker-3 ─── 连接C │ ...（16 个 worker） │ └── Redis 连接同理推论一：最大连接数 = worker 进程数 × 每进程连接数。 16 个 worker、每个用了 2 个连接配置（读写分离），就是 32 个常驻连接。上线前先算这笔账，别和 MySQL 的 max_connections 打架。推论二：连接是懒建立的。 进程启动时不连库，第一次查询才建立。所以刚 reload 完的进程有冷启动延迟，压测时记得先预热。推论三：连接会"死"。 MySQL wait_timeout（默认 8 小时）内没有活动，服务端单方面断开，webman 侧还握着死连接，下一个请求直接 MySQL has gone away。二、断线重连：三层配置兜底以 think-orm（webman 默认集成）为例：// config/thinkorm.php return [ 'default' => 'mysql', 'connections' => [ 'mysql' => [ 'type' => 'mysql', 'hostname' => '127.0.0.1', 'database' => 'app', 'username' => 'root', 'password' => 'root', 'charset' => 'utf8mb4', 'break_reconnect' => true, // ① 断线自动重连 'fields_cache' => true, // ② 字段信息缓存（常驻内存的红利） ], ], ];三层兜底缺一不可：break_reconnect => true：ORM 层检测断连自动重连心跳保活：定时进程每 5 分钟发一条 SELECT 1，让连接永不超时（webman 的 config/process.php 里加自定义进程实现）重试幂等：断线重连只对查询安全，写操作重试必须保证幂等，否则可能双写// 心跳进程示例 return [ 'db-keepalive' => [ 'handler' => function () { while (true) { Db::query('SELECT 1'); sleep(300); } }, ], ];三、ORM 选型：think-orm 还是 Laravel Eloquent？webman 官方两者都支持，选型看团队基因：维度think-ormEloquent（illuminate/database）上手成本ThinkPHP 系团队零成本Laravel 系团队零成本模型功能基础够用关联预加载、访问器、软删更丰富常驻内存适配fields_cache 等针对性优化需要自己注意静态缓存生态迁移TP 项目平移Laravel 项目平移中大型项目的建议：ORM 只做"查询构造 + 结果映射"，复杂查询直接裸 SQL 或查询构造器。模型层越薄，常驻内存下的心智负担越小。一个 Eloquent 用户必须知道的区别：FPM 下 DB::connection() 每请求新建，webman 下连接跨请求驻留——事务没有提交/回滚，下一个请求会接着在这个事务里跑，数据不一致的幽灵bug就是这么来的。think-orm 的 Db::startTrans() 同理。// ❌ 事故现场：异常没回滚，连接带着事务回到池里 try { Db::startTrans(); // ... 写操作 throw new \RuntimeException('boom'); // 没有 catch 回滚！ } finally { Db::commit(); // finally 里 commit，异常时也 commit 了半截数据？ }四、事务的正确姿势铁律一：事务边界必须完整闭合try { Db::startTrans(); $this->deductStock($skuId, $num); // 扣库存 $this->createOrder($uid, $skuId); // 建订单 Db::commit(); } catch (\Throwable $e) { Db::rollback(); throw $e; // 异常继续抛，别吞 }铁律二：事务里禁止外部 IO事务持锁期间调外部 HTTP / 发 MQ / 写 Redis，一旦外部服务慢，数据库连接和行锁被拖住，并发一上来就是雪崩。// ❌ 事务里调第三方风控接口（300ms），行锁拖 300ms Db::startTrans(); $stock = Stock::lock(true)->find($skuId); // SELECT ... FOR UPDATE RiskApi::check($uid); // 外部调用！ $stock->num -= 1; $stock->save(); Db::commit(); // ✅ 外部检查前置，事务只做纯 DB 操作 RiskApi::check($uid); // 先调外部 Db::startTrans(); $stock = Stock::lock(true)->find($skuId); if ($stock->num < 1) throw new OutOfStock(); $stock->num -= 1; $stock->save(); Db::commit(); PublishOrderEvent::dispatch($order); // 后置异步动作铁律三：事务尽量短，锁尽量准悲观锁（lock(true)）只锁必要的行，锁定前先校验数据存在的条件能用唯一索引防重（幂等）就别用"先查再插 + 锁"补充：嵌套事务与 savepointthink-orm 支持 startTrans 嵌套（底层转 savepoint），但事务边界只写在 Service 层一处，不要 Controller 包一层、Service 再包一层——嵌套事务的回滚语义很容易把人绕晕。五、大数据量：分批处理的三个工具1. chunk 分批读User::where('status', 1)->chunkById(1000, function ($users) { foreach ($users as $user) { // 逐批处理 } });用 chunkById 而不是 chunk：处理过程中如果更新了 offset 相关字段，chunk（基于 offset/limit）会漏数据或死循环，chunkById（基于主键游标）不会。2. 慎用 ORM 批量，必要时裸 SQLORM 逐条 save 在万级数据下慢得无法接受：// ❌ 10 万条逐条 update：几十分钟 foreach ($rows as $row) { Model::where('id', $row['id'])->update($row); } // ✅ 事务 + 批量拼接（注意分批，每批 ≤ 1000） Db::startTrans(); foreach (array_chunk($rows, 500) as $batch) { Db::table('users')->upsert($batch, ['id'], ['name', 'status']); // 批量 upsert } Db::commit();3. 超大导出：游标 + 流式输出百万行导出别 ORM ->get() 全量载入（内存直接爆），用游标逐行读、边读边写 CSV 流：$fp = fopen('php://output', 'w'); User::where('created', '>', $start)->select()->cursor()->each(function ($u) use ($fp) { fputcsv($fp, $u->toArray()); });cursor() 用 PDO 的 unbuffered 模式，内存占用恒定。六、读写分离与多库配置'connections' => [ 'mysql' => [ /* 主库写 */ ], 'mysql_read' => [ /* 从库读 */ ], ],think-orm/Eloquent 都支持配置读写分离自动路由，但强制主读（写后立读的场景：下单后立即查详情）必须显式指定主库连接，否则撞上主从延迟会查到旧数据：// 写后立刻要读的，强制走主库 $inDb = Order::on('mysql')->find($orderId);七、上线前检查清单[ ] 最大连接数核算：worker 数 × 每进程连接数 < MySQL max_connections × 0.8[ ] break_reconnect 开启 + 心跳进程就位[ ] 事务全部 try/catch/rollback 完整闭合（全局搜 startTrans 逐个人工审计）[ ] 长事务里无外部 IO（APM 看事务平均持锁时长）[ ] 大数据任务用 chunkById / cursor，无全量 ->get()[ ] ORM 字段缓存开启，生产关掉 SQL 日志写在最后数据库层的所有坑都指向同一个根因：连接的生命周期从"请求级"变成了"进程级"。把连接当进程级资源来管理（池化、心跳、重连），把事务当危险品来对待（短小、闭合、无 IO），webman 的数据库层就稳了。下一篇进入性能深水区：协程、连接池调优与压测。
Redis 缓存实战：穿透、击穿、雪崩三大问题解决方案
前言Redis 作为缓存使用时，会遇到三大经典问题：缓存穿透、缓存击穿和缓存雪崩。这三个问题在实际生产中频繁出现，处理不当可能导致数据库被压垮甚至服务不可用。本文将深入分析每个问题的成因并给出完整解决方案。一、缓存穿透1.1 问题描述用户请求的数据在缓存和数据库中都不存在，每次请求都直接打到数据库。常见于恶意攻击（用不存在的 ID 大量请求）或业务异常数据。请求 → 缓存MISS → 数据库MISS → 返回空 ↑ │ └──── 不断重复 ──────────────────────┘1.2 危害数据库承受大量无效查询如果是恶意攻击，可能打垮数据库浪费服务器资源1.3 解决方案方案一：缓存空值function getUser($id) { $cacheKey = "user:{$id}"; $data = $redis->get($cacheKey); if ($data !== false) { // 命中缓存（包括空值缓存） return $data === 'NULL' ? null : json_decode($data, true); } // 查数据库 $data = $db->query("SELECT * FROM users WHERE id = ?", [$id]); if (empty($data)) { // 缓存空值，设置较短过期时间（60秒） $redis->setex($cacheKey, 60, 'NULL'); return null; } $redis->setex($cacheKey, 3600, json_encode($data)); return $data; }优点：实现简单缺点：可能缓存大量空值，占用内存；数据后来新增了也不会被感知（除非过期）方案二：布隆过滤器（Bloom Filter）// 系统初始化时，将所有存在的ID加入布隆过滤器 function initBloomFilter() { $redis->del('bloom:user_ids'); $userIds = $db->query("SELECT id FROM users"); foreach ($userIds as $id) { // 使用多个哈希函数 bloomAdd('bloom:user_ids', $id); } } // 检查ID是否可能存在 function bloomAdd($key, $value) { $hashes = multiHash($value); foreach ($hashes as $h) { $redis->setBit($key, $h, 1); } } function bloomExists($key, $value) { $hashes = multiHash($value); foreach ($hashes as $h) { if ($redis->getBit($key, $h) == 0) { return false; // 一定不存在 } } return true; // 可能存在（有误判率） } // 请求处理 function getUser($id) { // 先过布隆过滤器 if (!bloomExists('bloom:user_ids', $id)) { return null; // 一定不存在，直接返回 } // 再查缓存和数据库 // ... }方案三：接口层参数校验// 在入口处拦截非法参数 function getUser($id) { // ID 必须是正整数 if (!is_numeric($id) || $id <= 0 || $id > PHP_INT_MAX) { throw new InvalidArgumentException('Invalid user ID'); } // 其他业务规则校验 if (strlen($id) > 10) { throw new InvalidArgumentException('ID too long'); } // 正常流程... }二、缓存击穿2.1 问题描述某个热点 key 突然过期（或被删除），大量并发请求同时访问这个 key，全部穿透到数据库。热点key过期 │ ├──→ 请求1 → 缓存MISS → 查DB ──┐ ├──→ 请求2 → 缓存MISS → 查DB ──┤ ├──→ 请求3 → 缓存MISS → 查DB ──┤ ├──→ 请求N → 缓存MISS → 查DB ──┤ │ │ └──────── 数据库被压垮 ◄────────┘2.2 解决方案方案一：互斥锁（推荐）function getHotData($key) { $data = $redis->get($key); if ($data !== false) { return json_decode($data, true); } // 获取互斥锁 $lockKey = "lock:{$key}"; $lockValue = uniqid('', true); // SET NX EX 尝试加锁 $locked = $redis->set($lockKey, $lockValue, ['NX', 'EX' => 10]); if ($locked) { try { // 再次检查缓存（双重检查） $data = $redis->get($key); if ($data !== false) { return json_decode($data, true); } // 查数据库 $data = $db->query("SELECT * FROM hot_table WHERE key_name = ?", [$key]); // 写入缓存 $redis->setex($key, 3600, json_encode($data)); return $data; } finally { // 释放锁（Lua脚本保证原子性） $lua = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end"; $redis->eval($lua, [$lockKey, $lockValue], 1); } } else { // 没拿到锁，短暂等待后重试 usleep(100000); // 100ms return getHotData($key); // 递归重试 } }方案二：逻辑过期不给 key 设置 TTL，而是在 value 中存储逻辑过期时间。后台异步刷新：function getWithLogicalExpire($key, $expireSeconds) { $data = $redis->get($key); if ($data === false) { return null; // 首次加载，需要手动初始化 } $cached = json_decode($data, true); $now = time(); // 未逻辑过期，直接返回 if ($cached['expire_at'] > $now) { return $cached['data']; } // 已逻辑过期，尝试获取锁刷新 $lockKey = "lock:refresh:{$key}"; $locked = $redis->set($lockKey, '1', ['NX', 'EX' => 10]); if ($locked) { // 后台异步刷新（非阻塞） // 实际项目中可以用队列或协程 go(function() use ($key, $expireSeconds) { $data = $db->query("SELECT * FROM table WHERE key_name = ?", [$key]); $redis->set($key, json_encode([ 'data' => $data, 'expire_at' => time() + $expireSeconds, ])); $redis->del("lock:refresh:{$key}"); }); } // 无论是否刷新，先返回旧数据（不阻塞用户） return $cached['data']; }方案三：永不过期 + 主动更新// 热点数据不设置过期时间 $redis->set('hot:ranking', json_encode($data)); // 不设置 TTL // 数据变更时主动更新缓存 function updateRanking() { $data = $db->query("SELECT * FROM ranking ORDER BY score DESC LIMIT 100"); $redis->set('hot:ranking', json_encode($data)); // 仍然不设置 TTL } // 定时任务刷新 // crontab: */10 * * * * php artisan ranking:refresh三、缓存雪崩3.1 问题描述大量 key 在同一时间集中过期，或者 Redis 宕机，导致大量请求同时打到数据库。大量key同时过期 │ ├──→ 请求1 → MISS → 查DB ──┐ ├──→ 请求2 → MISS → 查DB ──┤ ├──→ 请求3 → MISS → 查DB ──┤ ├──→ 请求N → MISS → 查DB ──┤ │ │ └──── 数据库被压垮 ◄────────┘3.2 与缓存击穿的区别对比缓存击穿缓存雪崩影响范围单个热点 key大量 key原因热点 key 过期大量 key 同时过期 / Redis 宕机量级小大3.3 解决方案方案一：随机过期时间function cacheData($key, $data, $baseTTL = 3600) { // 在基础TTL上加随机偏移，避免同时过期 $randomTTL = $baseTTL + mt_rand(0, 600); // 0~10分钟随机 $redis->setex($key, $randomTTL, json_encode($data)); } // 批量缓存时尤其重要 function batchCache($items) { foreach ($items as $key => $value) { $ttl = 3600 + mt_rand(0, 1800); // 1~1.5小时随机 $redis->setex($key, $ttl, json_encode($value)); } }方案二：多级缓存function getWithMultiLevel($key) { // L1: 本地缓存（进程内） $data = $localCache->get($key); if ($data !== null) { return $data; } // L2: Redis 缓存 $data = $redis->get($key); if ($data !== false) { $localCache->set($key, $data, 60); // 本地缓存60秒 return json_decode($data, true); } // L3: 数据库 $data = $db->query("SELECT * FROM table WHERE key_name = ?", [$key]); if ($data) { $redis->setex($key, 3600 + mt_rand(0, 600), json_encode($data)); $localCache->set($key, $data, 60); } return $data; }方案三：熔断降级function getProductList($categoryId) { try { $data = $redis->get("products:{$categoryId}"); if ($data !== false) { return json_decode($data, true); } // 数据库查询带超时和限流 $data = $db->query("SELECT * FROM products WHERE category_id = ?", [$categoryId], ['timeout' => 1]); $redis->setex("products:{$categoryId}", 3600 + mt_rand(0, 600), json_encode($data)); return $data; } catch (Exception $e) { // 降级：返回默认数据或静态数据 return getDefaultProductList(); } }方案四：Redis 高可用# Redis Sentinel 哨兵模式自动故障转移 sentinel monitor mymaster 192.168.1.10 6379 2 sentinel down-after-milliseconds mymaster 30000 sentinel failover-timeout mymaster 180000 # Redis Cluster 集群模式 # 数据分散在多个节点，单节点宕机不影响整体四、综合对比问题本质核心方案推荐组合缓存穿透查询不存在数据布隆过滤器 + 空值缓存布隆过滤器 + 短TTL空值缓存击穿热点key过期互斥锁 + 双重检查互斥锁 + 逻辑过期缓存雪崩大量key同时过期随机TTL + 多级缓存随机TTL + 熔断降级五、完整防护代码示例class CacheService { private $redis; private $db; private $localCache = []; /** * 安全缓存读取（防穿透+击穿+雪崩） */ public function get($key, $dbQuery, $ttl = 3600) { // 1. 本地缓存 if (isset($this->localCache[$key])) { return $this->localCache[$key]; } // 2. Redis 缓存 $data = $this->redis->get($key); if ($data !== false) { if ($data === 'NULL') return null; $this->localCache[$key] = json_decode($data, true); return $this->localCache[$key]; } // 3. 互斥锁防击穿 $lockKey = "lock:{$key}"; $locked = $this->redis->set($lockKey, '1', ['NX', 'EX' => 10]); if (!$locked) { usleep(100000); // 等待100ms return $this->get($key, $dbQuery, $ttl); // 重试 } try { // 双重检查 $data = $this->redis->get($key); if ($data !== false) { return $data === 'NULL' ? null : json_decode($data, true); } // 4. 查数据库 $data = $dbQuery(); // 5. 缓存（随机TTL防雪崩） $randomTTL = $ttl + mt_rand(0, 300); if ($data === null) { // 空值缓存（短TTL防穿透） $this->redis->setex($key, 60, 'NULL'); } else { $this->redis->setex($key, $randomTTL, json_encode($data)); $this->localCache[$key] = $data; } return $data; } finally { $this->redis->del($lockKey); } } } // 使用 $cache = new CacheService(); $user = $cache->get("user:1001", function() use ($db) { return $db->query("SELECT * FROM users WHERE id = 1001"); }, 3600);总结缓存三大问题的核心防护策略：穿透：布隆过滤器拦截 + 空值缓存兜底击穿：互斥锁保证只有一个请求查DB + 双重检查雪崩：随机TTL打散过期时间 + 多级缓存 + 熔断降级生产环境建议三者组合使用，形成完整的缓存防护体系。
前言Redis 是目前最流行的内存数据库，凭借极高的读写性能和丰富的数据结构，被广泛应用于缓存、消息队列、分布式锁等场景。本文将系统讲解 Redis 的五大核心数据类型及其常用命令，帮助你快速上手。一、Redis 数据类型总览数据类型底层结构典型应用场景StringSDS（简单动态字符串）缓存、计数器、分布式锁Hashziplist / hashtable对象存储、用户信息Listquicklist（ziplist + 链表）消息队列、文章列表Setintset / hashtable标签、共同好友、去重ZSetziplist / skiplist + hashtable排行榜、延迟队列二、String（字符串）String 是 Redis 最基础的类型，一个 key 对应一个 value，最大支持 512MB。2.1 基本命令# 设置值 SET name "redis" SET name "redis" EX 30 # 30秒过期 SET name "redis" PX 30000 # 30000毫秒过期 SET name "redis" NX # 仅当key不存在时设置 SET name "redis" XX # 仅当key存在时设置 # 获取值 GET name # "redis" # 批量操作 MSET k1 "v1" k2 "v2" k3 "v3" MGET k1 k2 k3 # 1) "v1" 2) "v2" 3) "v3" # 追加 APPEND name " tutorial" # "redis tutorial" # 长度 STRLEN name # 14 # 自增自减（值必须是数字） SET counter 100 INCR counter # 101 INCRBY counter 10 # 111 DECR counter # 110 DECRBY counter 20 # 902.2 实战：页面访问计数器# 每次访问页面自增 INCR page:home:views # 1 INCR page:home:views # 2 INCR page:home:views # 3 # 获取今日访问量（按日期分key） INCR page:home:views:20260724 # 1 # 设置过期时间，自动清理历史数据 EXPIRE page:home:views:20260724 864002.3 实战：分布式锁（简单版）# 加锁：SET NX 保证原子性 SET lock:order:1001 "uuid-xxx" NX EX 30 # 释放锁（需要Lua脚本保证原子性） # eval "if redis.call('get',KEYS[1])==ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end" 1 lock:order:1001 uuid-xxx三、Hash（哈希）Hash 是一个键值对集合，特别适合存储对象。3.1 基本命令# 设置字段 HSET user:1001 name "张三" HSET user:1001 age 25 HSET user:1001 email "zhangsan@example.com" # 批量设置 HMSET user:1002 name "李四" age 30 email "lisi@example.com" # 获取字段 HGET user:1001 name # "张三" HMGET user:1001 name age # 1) "张三" 2) "25" # 获取所有字段 HGETALL user:1001 # 1) "name" # 2) "张三" # 3) "age" # 4) "25" # 5) "email" # 6) "zhangsan@example.com" # 获取所有字段名/值 HKEYS user:1001 # name, age, email HVALS user:1001 # 张三, 25, zhangsan@example.com # 字段数量 HLEN user:1001 # 3 # 删除字段 HDEL user:1001 email # 字段自增 HINCRBY user:1001 age 1 # 263.2 实战：购物车# 添加商品到购物车（user:1001 的购物车） HSET cart:1001 product:2001 2 # 商品2001，数量2 HSET cart:1001 product:2002 1 # 商品2002，数量1 # 修改数量 HINCRBY cart:1001 product:2001 1 # 数量+1 → 3 # 查看购物车 HGETALL cart:1001 # 移除商品 HDEL cart:1001 product:2002 # 清空购物车 DEL cart:1001四、List（列表）List 是有序的字符串列表，支持在头部或尾部插入/弹出元素。4.1 基本命令# 左/右推入 LPUSH messages "msg1" "msg2" # 列表: msg2, msg1 RPUSH messages "msg3" # 列表: msg2, msg1, msg3 # 左/右弹出 LPOP messages # "msg2" RPOP messages # "msg3" # 查看范围 LRANGE messages 0 -1 # 所有元素 LRANGE messages 0 2 # 前3个 # 长度 LLEN messages # 按索引获取 LINDEX messages 0 # 第一个元素 # 按索引设置 LSET messages 0 "new_msg" # 阻塞弹出（消息队列核心命令） BLPOP queue:task 30 # 阻塞最多30秒 BRPOP queue:task 304.2 实战：简单消息队列# 生产者：推入任务 LPUSH queue:email '{"to":"a@b.com","subject":"Hello"}' # 消费者：阻塞式弹出任务 BRPOP queue:email 0 # 0表示永久阻塞 # 优点：简单、原子性 # 缺点：无ACK机制，消息丢失需自行处理五、Set（集合）Set 是无序的、不重复的字符串集合。5.1 基本命令# 添加成员 SADD tags:article:1 "redis" "database" "cache" # 查看所有成员 SMEMBERS tags:article:1 # redis, database, cache # 是否是成员 SISMEMBER tags:article:1 "redis" # 1（true） # 成员数量 SCARD tags:article:1 # 3 # 移除成员 SREM tags:article:1 "cache" # 随机弹出 SPOP tags:article:1 # 随机移除并返回一个 SRANDMEMBER tags:article:1 2 # 随机返回2个（不移除）5.2 集合运算SADD set:a "1" "2" "3" "4" SADD set:b "3" "4" "5" "6" # 交集 SINTER set:a set:b # 3, 4 # 并集 SUNION set:a set:b # 1, 2, 3, 4, 5, 6 # 差集（a有但b没有） SDIFF set:a set:b # 1, 25.3 实战：共同关注# 用户A关注了 SADD follows:user:A "redis" "mysql" "docker" # 用户B关注了 SADD follows:user:B "redis" "nginx" "docker" # 共同关注 SINTER follows:user:A follows:user:B # redis, docker # A关注但B没关注 SDIFF follows:user:A follows:user:B # mysql # 推荐关注（B关注但A没关注） SDIFF follows:user:B follows:user:A # nginx六、ZSet（有序集合）ZSet 每个成员关联一个分数（score），按分数排序，是 Redis 最强大的数据结构之一。6.1 基本命令# 添加成员（带分数） ZADD ranking 100 "player1" ZADD ranking 200 "player2" ZADD ranking 150 "player3" ZADD ranking 300 "player4" # 按分数升序排名（0开始） ZRANGE ranking 0 -1 # player1, player3, player2, player4 ZRANGE ranking 0 -1 WITHSCORES # 带分数 # 按分数降序排名 ZREVRANGE ranking 0 -1 # player4, player2, player3, player1 ZREVRANGE ranking 0 -1 WITHSCORES # 按分数范围查询 ZRANGEBYSCORE ranking 100 200 # 100~200分之间 ZRANGEBYSCORE ranking "(100" 200 # 不包含100 # 获取成员分数 ZSCORE ranking "player2" # 200 # 获取成员排名 ZRANK ranking "player2" # 2（升序第3名，0开始） ZREVRANK ranking "player2" # 1（降序第2名） # 增加分数 ZINCRBY ranking 50 "player1" # player1分数变为150 # 成员数量 ZCARD ranking # 4 # 分数范围内的成员数 ZCOUNT ranking 100 200 # 36.2 实战：游戏排行榜# 玩家得分 ZADD game_rank 1500 "Alice" ZADD game_rank 2300 "Bob" ZADD game_rank 1800 "Charlie" ZADD game_rank 2100 "David" ZADD game_rank 1900 "Eve" # Top 3 ZREVRANGE game_rank 0 2 WITHSCORES # 1) "Bob" 2) "2300" # 3) "David" 4) "2100" # 5) "Eve" 6) "1900" # 查看某玩家排名 ZREVRANK game_rank "Alice" # 4（第5名） # 玩家加分 ZINCRBY game_rank 100 "Alice" # Alice: 1600 # 分数段统计 ZCOUNT game_rank 1500 2000 # 3人（Alice, Charlie, Eve）6.3 实战：延迟队列# 将任务加入延迟队列（score为执行时间戳） ZADD delay_queue 1721800500 "task1" # 5秒后执行 ZADD delay_queue 1721800600 "task2" # 15秒后执行 ZADD delay_queue 1721800700 "task3" # 25秒后执行 # 消费者轮询：取出已到期的任务 # 用 Lua 脚本保证原子性 # eval "local items = redis.call('ZRANGEBYSCORE', KEYS[1], 0, ARGV[1]) # if #items > 0 then # redis.call('ZREMRANGEBYSCORE', KEYS[1], 0, ARGV[1]) # end # return items" 1 delay_queue <当前时间戳>七、其他常用命令7.1 Key 管理# 查找key（生产环境慎用 KEYS，用 SCAN） KEYS user:* # 所有以user:开头的key SCAN 0 MATCH user:* COUNT 100 # 渐进式扫描 # 过期时间 EXPIRE name 30 # 30秒后过期 EXPIREAT name 1721800000 # 在指定时间戳过期 TTL name # 查看剩余秒数（-1永久，-2已过期） PERSIST name # 移除过期时间 # 类型判断 TYPE name # string TYPE user:1001 # hash7.2 通用操作# 删除 DEL name # 删除单个 DEL name1 name2 # 删除多个 # 判断存在 EXISTS name # 1（存在） # 重命名 RENAME name newname总结类型核心特点一句话记忆String一对一万能缓存Hash一对多对象存储List有序可重复消息队列Set无序不重复去重与运算ZSet有序不重复排行榜掌握这五大类型和核心命令，就能应对 90% 以上的 Redis 使用场景。后续文章将深入持久化、缓存策略和集群部署。
Redis 持久化机制深度对比：RDB、AOF 与混合持久化
前言Redis 作为内存数据库，数据存储在 RAM 中，一旦服务器重启数据就会丢失。为了解决这个问题，Redis 提供了两种持久化机制：RDB（快照）和 AOF（追加日志）。本文将深入对比这两种机制的原理、配置和选择策略。一、RDB（Redis Database）1.1 工作原理RDB 是将 Redis 内存中的数据以二进制快照的形式写入磁盘。触发方式有三种：┌─────────────────────────────────────┐ │ Redis 内存数据 │ │ │ │ ┌───────┐ ┌───────┐ │ │ │ Key1 │ │ Key2 │ ... │ │ └───────┘ └───────┘ │ │ │ │ ▼ fork() 子进程 │ │ │ │ ┌─────────────────────┐ │ │ │ BGSAVE 子进程 │ │ │ │ 写入 dump.rdb │ │ │ └─────────────────────┘ │ └─────────────────────────────────────┘1.2 触发方式手动触发：# 阻塞式（生产环境禁用） SAVE # 后台异步执行（推荐） BGSAVE自动触发（配置文件）：# redis.conf # 900秒内至少1个key变化 → 触发RDB save 900 1 # 300秒内至少10个key变化 → 触发RDB save 300 10 # 60秒内至少10000个key变化 → 触发RDB save 60 10000 # 禁用自动RDB # save ""其他触发场景：主从复制时，主节点自动触发 BGSAVE执行 SHUTDOWN 命令时执行 FLUSHALL 命令时（生成空快照）1.3 RDB 文件配置# redis.conf # RDB 文件名 dbfilename dump.rdb # 存储目录 dir /var/lib/redis # 压缩（默认开启，用 LZF 压缩） rdbcompression yes # 校验和（默认开启） rdbchecksum yes # BGSAVE 出错时停止写入（默认yes，推荐开启） stop-writes-on-bgsave-error yes1.4 RDB 优缺点优点：文件紧凑，体积小，适合备份和传输恢复速度快（直接加载二进制数据）对性能影响小（子进程异步执行）适合做冷备和灾难恢复缺点：宕机时会丢失最后一次快照后的所有数据fork 子进程时，如果数据量大，可能造成短暂阻塞不适合对数据完整性要求高的场景二、AOF（Append Only File）2.1 工作原理AOF 以追加的方式将每条写命令记录到日志文件中，重启时按顺序回放命令恢复数据。┌─────────────────────────────────────┐ │ Redis 内存数据 │ │ │ │ 客户端写入命令 │ │ │ │ │ ▼ │ │ ┌──────────┐ ┌──────────────┐ │ │ │ Redis执行 │ → │ 追加到AOF缓冲 │ │ │ │ 命令 │ │ 区 │ │ │ └──────────┘ └──────┬───────┘ │ │ │ │ │ ▼ fsync │ │ ┌──────┴───────┐ │ │ │ appendonly.aof│ │ │ └──────────────┘ │ └─────────────────────────────────────┘2.2 开启 AOF# redis.conf appendonly yes appendfilename "appendonly.aof"2.3 刷盘策略# 三种 fsync 策略 # always: 每条命令都fsync（最安全，性能最差） # everysec: 每秒fsync一次（推荐，最多丢1秒数据） # no: 由OS决定何时fsync（性能最好，可能丢较多数据） appendfsync everysec策略数据安全性性能适用场景always最高（不丢数据）最差对数据极其敏感everysec高（最多丢1秒）好推荐默认no低最好纯缓存场景2.4 AOF 重写随着时间推移，AOF 文件会越来越大。Redis 通过"重写"机制压缩文件：分析当前内存数据，用最少的命令重新生成 AOF 文件。# 手动触发重写 BGREWRITEAOF# redis.conf # AOF 文件大小比上次重写后增长 100% 时自动重写 auto-aof-rewrite-percentage 100 # AOF 文件最小重写体积（64MB） auto-aof-rewrite-min-size 64mb # 重写期间不执行 fsync（减少磁盘 IO） no-appendfsync-on-rewrite yes重写过程：原始 AOF: SET key1 "v1" SET key1 "v2" ← 覆盖 DEL key1 ← 删除 SET key1 "v3" ← 最终值 重写后: SET key1 "v3" ← 只保留最终状态2.5 AOF 优缺点优点：数据安全性高（最多丢 1 秒数据）日志格式可读，可手动修复支持实时持久化，适合对数据完整性要求高的场景缺点：文件体积比 RDB 大恢复速度比 RDB 慢（需要回放命令）重写时会 fork 子进程，大数据量时可能阻塞三、RDB + AOF 混合持久化Redis 4.0 引入混合持久化，结合两者的优点：# redis.conf aof-use-rdb-preamble yes工作原理：AOF 重写时，先以 RDB 格式写入当前内存快照重写期间的增量写命令以 AOF 格式追加在 RDB 数据之后┌────────────────────────────────────┐ │ 混合 AOF 文件 │ │ │ │ ┌──────────────────────────────┐ │ │ │ RDB 二进制数据（全量快照） │ │ │ │ - 恢复速度快 │ │ │ │ - 文件体积小 │ │ │ └──────────────────────────────┘ │ │ ┌──────────────────────────────┐ │ │ │ AOF 增量命令（重写期间） │ │ │ │ - 保证数据不丢 │ │ │ └──────────────────────────────┘ │ └────────────────────────────────────┘恢复流程：先加载 RDB 部分（快速恢复大部分数据）再回放 AOF 增量命令（补全重写期间的数据）四、完整配置示例# redis.conf 生产环境推荐配置 # ===== RDB 配置 ===== save 900 1 save 300 10 save 60 10000 stop-writes-on-bgsave-error yes rdbcompression yes rdbchecksum yes dbfilename dump.rdb dir /var/lib/redis # ===== AOF 配置 ===== appendonly yes appendfilename "appendonly.aof" appendfsync everysec # 混合持久化（Redis 4.0+） aof-use-rdb-preamble yes # AOF 重写 auto-aof-rewrite-percentage 100 auto-aof-rewrite-min-size 64mb no-appendfsync-on-rewrite yes五、持久化方案选择场景推荐方案理由纯缓存，可接受数据丢失仅 RDB性能最优，恢复快数据库，不可丢数据仅 AOF数据安全性高综合场景（推荐）混合持久化兼顾性能和安全灾备备份RDB + 定时备份文件小，传输快备份策略建议#!/bin/bash # 每日凌晨2点备份 RDB 文件 BACKUP_DIR="/backup/redis/$(date +%Y%m%d)" mkdir -p $BACKUP_DIR # 触发 BGSAVE redis-cli BGSAVE # 等待完成 while [ $(redis-cli INFO persistence | grep rdb_bgsave_in_progress | tr -d '\r') != "0" ]; do sleep 1 done # 复制文件 cp /var/lib/redis/dump.rdb $BACKUP_DIR/ # 保留最近7天 find /backup/redis/ -type d -mtime +7 -exec rm -rf {} \; echo "Backup completed: $BACKUP_DIR/dump.rdb"六、故障恢复AOF 文件损坏修复# 如果 AOF 文件损坏，Redis 无法启动 redis-check-aof --fix appendonly.aof # 检查 RDB 文件 redis-check-rdb dump.rdb从备份恢复# 1. 停止 Redis systemctl stop redis # 2. 替换 RDB 文件 cp /backup/redis/20260724/dump.rdb /var/lib/redis/ # 3. 如果不想用 AOF，临时关闭 # 修改 redis.conf: appendonly no # 4. 启动 Redis systemctl start redis # 5. 验证数据 redis-cli DBSIZE总结对比项RDBAOF混合持久化数据安全可能丢分钟级数据最多丢1秒最多丢1秒文件大小小大中恢复速度快慢快性能影响小（fork时短暂）中（每秒fsync）中推荐场景备份/缓存数据库生产推荐生产环境推荐开启混合持久化，同时定期备份 RDB 文件做异地灾备。
前言当单机 Redis 无法满足业务需求时，就需要引入集群方案。Redis 提供了三种主要的集群模式：主从复制、哨兵模式和 Cluster 集群。本文将系统讲解三种方案的原理、搭建和适用场景。一、主从复制1.1 架构 ┌──────────────┐ │ Master │ │ (读+写) │ └──────┬───────┘ │ 复制 ┌──────────┼──────────┐ │ │ │ ┌──────▼───┐ ┌───▼────┐ ┌──▼─────┐ │ Slave 1 │ │Slave 2 │ │Slave 3 │ │ (只读) │ │(只读) │ │(只读) │ └──────────┘ └────────┘ └────────┘1.2 工作原理全量同步：Slave 首次连接 Master 时，Master 执行 BGSAVE 生成 RDB 文件，发送给 Slave增量同步：全量同步完成后，Master 将新的写命令实时同步给 Slave断线重连：Slave 断线后重连，通过 replid + offset 判断是否可以增量同步1.3 搭建配置# slave1.conf port 6380 # 配置主节点 slaveof 192.168.1.10 6379 # 从节点只读 slave-read-only yes # 主节点密码 masterauth your_password# 启动从节点 redis-server slave1.conf # 验证主从关系 redis-cli -p 6380 INFO replication # role:slave # master_host:192.168.1.10 # master_port:63791.4 优缺点优点缺点配置简单Master 宕机需手动切换读写分离，提升读性能不具备自动故障转移数据有副本全量同步时可能阻塞 Master二、哨兵模式（Sentinel）2.1 架构┌──────────┐ ┌──────────┐ ┌──────────┐ │Sentinel 1│ │Sentinel 2│ │Sentinel 3│ │ (监控) │ │ (监控) │ │ (监控) │ └────┬─────┘ └────┬─────┘ └────┬─────┘ │ │ │ └──────┬───────┴──────┬──────┘ │ │ ┌──────▼──────┐ ┌────▼─────┐ │ Master │ │ Slave │ │ (读+写) │ │ (只读) │ └─────────────┘ └──────────┘2.2 工作原理监控：Sentinel 不断向 Master/Slave 发送 PING 命令主观下线：某个 Sentinel 发现节点无响应，标记为主观下线客观下线：超过半数 Sentinel 都报告节点下线，标记为客观下线选举 Leader：Sentinel 之间选举一个 Leader 执行故障转移故障转移：Leader 选一个 Slave 提升为新 Master，通知其他 Slave 和客户端2.3 搭建配置# sentinel1.conf port 26379 # 监控的 Master（名称 IP 端口 仲裁人数） sentinel monitor mymaster 192.168.1.10 6379 2 # Master 无响应 30 秒判定下线 sentinel down-after-milliseconds mymaster 30000 # 故障转移超时时间 sentinel failover-timeout mymaster 180000 # 同时只允许1个Slave做同步 sentinel parallel-syncs mymaster 1 # Master 密码 sentinel auth-pass mymaster your_password# 启动3个Sentinel redis-sentinel sentinel1.conf redis-sentinel sentinel2.conf redis-sentinel sentinel3.conf2.4 故障转移过程1. Master宕机 │ 2. Sentinels检测到Master无响应 │ 3. 超过quorum(2)个Sentinel标记为客观下线 │ 4. 选举Leader Sentinel │ 5. Leader选择最优Slave提升为Master: - 排除故障Slave - 选择offset最大的Slave（数据最新） - 优先级高的Slave优先 │ 6. 其他Slave指向新Master │ 7. 通知客户端新Master地址 │ 8. 旧Master恢复后变为Slave2.5 客户端连接// PHP 客户端连接 Sentinel $sentinels = [ 'tcp://192.168.1.10:26379', 'tcp://192.168.1.11:26379', 'tcp://192.168.1.12:26379', ]; $redis = new RedisSentinel($sentinels, [ 'name' => 'mymaster', 'password' => 'your_password', ]); // 自动路由到当前 Master $redis->set('key', 'value');三、Cluster 集群3.1 架构┌─────────────────────────────────────────────────────┐ │ Redis Cluster │ │ │ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │ │ Node A │ │ Node B │ │ Node C │ │ │ │Master │ │Master │ │Master │ │ │ │槽0-5460 │ │槽5461- │ │槽10923- │ │ │ │ │ │ 10922 │ │ 16383 │ │ │ └────┬────┘ └────┬────┘ └────┬────┘ │ │ │ │ │ │ │ ┌────▼────┐ ┌────▼────┐ ┌────▼────┐ │ │ │ Node A' │ │ Node B' │ │ Node C' │ │ │ │ Slave │ │ Slave │ │ Slave │ │ │ └─────────┘ └─────────┘ └─────────┘ │ └─────────────────────────────────────────────────────┘3.2 核心概念哈希槽（Hash Slot）Redis Cluster 有 16384 个哈希槽，分布在所有 Master 节点上：# 计算key属于哪个槽 slot = CRC16(key) % 16384 # 示例 # key="user:1001" → CRC16 → slot=12345 # slot 12345 属于 Node C (10923-16383)节点通信（Gossip协议）每个节点定期向其他节点发送 PING 消息，交换集群状态信息。3.3 搭建 Cluster方式一：手动搭建（6节点，3主3从）# 创建6个配置文件 for port in 7001 7002 7003 7004 7005 7006; do mkdir -p /redis/cluster/$port cat > /redis/cluster/$port/redis.conf << EOF port $port cluster-enabled yes cluster-config-file nodes.conf cluster-node-timeout 5000 appendonly yes dir /redis/cluster/$port EOF done # 启动6个节点 for port in 7001 7002 7003 7004 7005 7006; do redis-server /redis/cluster/$port/redis.conf & done # 创建集群 redis-cli --cluster create \ 127.0.0.1:7001 127.0.0.1:7002 127.0.0.1:7003 \ 127.0.0.1:7004 127.0.0.1:7005 127.0.0.1:7006 \ --cluster-replicas 1方式二：Docker Compose 搭建version: '3.8' services: redis-node-1: image: redis:7 command: redis-server --cluster-enabled yes --cluster-config-file nodes.conf --cluster-node-timeout 5000 --appendonly yes --port 7001 ports: - "7001:7001" networks: - redis-cluster redis-node-2: image: redis:7 command: redis-server --cluster-enabled yes --cluster-config-file nodes.conf --cluster-node-timeout 5000 --appendonly yes --port 7002 ports: - "7002:7002" networks: - redis-cluster # ... node-3 到 node-6 同理 redis-cluster-init: image: redis:7 command: > bash -c "sleep 10 && echo yes | redis-cli --cluster create redis-node-1:7001 redis-node-2:7002 redis-node-3:7003 redis-node-4:7004 redis-node-5:7005 redis-node-6:7006 --cluster-replicas 1" depends_on: - redis-node-1 - redis-node-2 - redis-node-3 - redis-node-4 - redis-node-5 - redis-node-6 networks: - redis-cluster networks: redis-cluster: driver: bridge3.4 客户端使用// PHP 使用 Redis Cluster $redis = new RedisCluster( null, // null 表示自动发现节点 [ '127.0.0.1:7001', '127.0.0.1:7002', '127.0.0.1:7003', ], 2, // 连接超时 2, // 读写超时 true, // 持久化连接 'password' ); // 使用方式与单机一致，客户端自动路由 $redis->set('user:1001', 'Alice'); $redis->set('user:1002', 'Bob'); $redis->set('user:1003', 'Charlie'); // 不同的key可能落在不同节点 // user:1001 → slot 12345 → Node C // user:1002 → slot 6789 → Node B // user:1003 → slot 456 → Node A3.5 Hash Tag如果需要让多个 key 落在同一个槽，使用 Hash Tag：# 用 {} 指定hash计算的部分 SET {user:1001}:profile "Alice's profile" SET {user:1001}:settings '{"theme":"dark"}' SET {user:1001}:cart "[]" # 三个key都计算 user:1001 的CRC16，落在同一个槽 # 可以在同一节点执行批量操作 MGET {user:1001}:profile {user:1001}:settings {user:1001}:cart四、三种方案对比特性主从复制Sentinel 哨兵Cluster 集群数据分片否否是（16384槽）自动故障转移否是是读写分离是是是Master单点瓶颈是是否最少节点数23+Sentinel6复杂度低中高适用场景读多写少中小规模大规模五、选型建议数据量 < 10GB? ├── 是 → 单机Redis够用 │ └── 需要高可用? │ ├── 是 → Sentinel（3节点+3哨兵） │ └── 否 → 单机 │ └── 否 → 数据量大/并发高 └── Redis Cluster（3主3从起步）生产环境推荐中小项目：Sentinel 哨兵（1主2从+3哨兵）大型项目：Redis Cluster（3主3从起步，按需扩展）极大规模：Cluster + 多机房部署 + 代理层六、运维注意事项6.1 内存监控# 查看各节点内存使用 redis-cli -p 7001 INFO memory | grep used_memory_human redis-cli -p 7002 INFO memory | grep used_memory_human redis-cli -p 7003 INFO memory | grep used_memory_human # 集群整体状态 redis-cli -p 7001 CLUSTER INFO redis-cli -p 7001 CLUSTER NODES6.2 扩容（Cluster）# 新增节点 redis-server /redis/cluster/7007/redis.conf # 加入集群 redis-cli --cluster add-node 127.0.0.1:7007 127.0.0.1:7001 # 分配槽位（从现有节点迁移部分槽到新节点） redis-cli --cluster reshard 127.0.0.1:7001 \ --cluster-from <现有节点ID> \ --cluster-to <新节点ID> \ --cluster-slots 1000 \ --cluster-yes6.3 数据迁移# 在线迁移（不停服） redis-cli --cluster rebalance 127.0.0.1:7001 \ --cluster-use-empty-masters \ --cluster-threshold 1总结方案核心价值关键限制主从复制读写分离、数据备份无自动故障转移Sentinel自动故障转移、高可用单Master写入瓶颈Cluster数据分片、水平扩展不支持跨槽事务、复杂度高选择原则：先评估数据量和并发量，小规模用 Sentinel，大规模用 Cluster，不要过度设计。
前言分布式锁是分布式系统中控制资源访问的核心机制。Redis 因其单线程特性和高性能，是实现分布式锁的首选方案。本文将从最简单的实现出发，逐步演进到生产级的 Redlock 算法。一、为什么需要分布式锁1.1 单机锁的问题// 单机环境：PHP文件锁或flock可以解决 $fp = fopen('/tmp/lock.txt', 'w+'); flock($fp, LOCK_EX); // 加锁 // 执行业务逻辑 flock($fp, LOCK_UN); // 释放锁 fclose($fp);但在分布式环境中，请求可能被分配到不同服务器，文件锁无法跨机器生效。1.2 分布式锁的核心要求要求说明互斥性同一时刻只有一个客户端持有锁避免死锁锁必须有过期时间，防止持有者宕机导致永久锁死解铃还须系铃人只有锁的持有者才能释放锁高可用锁服务自身不能成为单点二、Redis 分布式锁演进2.1 版本一：SETNX（初版）// 最简单的实现 $lockKey = 'lock:order:1001'; // 尝试加锁 $redis->setnx($lockKey, 1); // 返回1表示成功，0表示失败 // 设置过期时间（防死锁） $redis->expire($lockKey, 30); // 业务逻辑 processOrder(); // 释放锁 $redis->del($lockKey);致命问题：setnx 和 expire 是两条命令，不保证原子性。如果 setnx 成功后服务崩溃，锁将永远不会过期，导致死锁。2.2 版本二：SET NX EX（推荐基础版）$lockKey = 'lock:order:1001'; $lockValue = uniqid('', true); // 生成唯一标识 // 一条命令完成加锁+过期，原子操作 $result = $redis->set($lockKey, $lockValue, ['NX', 'EX' => 30]); if ($result) { try { // 获取锁成功，执行业务 processOrder(); } finally { // 释放锁 $redis->del($lockKey); } } else { // 获取锁失败 throw new Exception('系统繁忙，请稍后重试'); }改进：SET key value NX EX 30 是原子操作，解决了死锁问题。遗留问题：释放锁时直接 DEL，可能删掉别人的锁。2.3 版本三：锁标识 + Lua 释放（安全版）$lockKey = 'lock:order:1001'; $lockValue = uniqid('', true); // 加锁 $result = $redis->set($lockKey, $lockValue, ['NX', 'EX' => 30]); if (!$result) { throw new Exception('获取锁失败'); } try { processOrder(); } finally { // 用Lua脚本安全释放锁：只有value匹配才删除 $luaScript = " if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end "; $redis->eval($luaScript, [$lockKey, $lockValue], 1); }为什么需要 Lua 脚本？时间线： T1: 客户端A GET lock → value=A T2: 锁过期，自动删除 T3: 客户端B SET lock NX → 成功，value=B T4: 客户端A DEL lock → 删掉了B的锁！Lua 脚本保证了 GET + 判断 + DEL 三步原子执行。2.4 版本四：锁续期（生产可用版）业务执行时间可能超过锁的过期时间，需要后台续期：class RedisLock { private $redis; private $lockKey; private $lockValue; private $ttl = 30; private $renewTimer = null; public function lock($key, $ttl = 30) { $this->lockKey = $key; $this->lockValue = uniqid('', true); $this->ttl = $ttl; // 尝试加锁，最多重试3次 for ($i = 0; $i < 3; $i++) { $result = $this->redis->set( $this->lockKey, $this->lockValue, ['NX', 'EX' => $this->ttl] ); if ($result) { $this->startRenew(); return true; } usleep(100000); // 100ms后重试 } return false; } /** * 后台续期（守护进程/定时器） * 每隔 TTL/3 秒续期一次 */ private function startRenew() { $renewInterval = intval($this->ttl / 3); // 使用Swoole定时器（或PCNTL信号） if (function_exists('swoole_timer_tick')) { $this->renewTimer = swoole_timer_tick( $renewInterval * 1000, function() { $this->renew(); } ); } } private function renew() { $luaScript = " if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('expire', KEYS[1], ARGV[2]) else return 0 end "; $this->redis->eval( $luaScript, [$this->lockKey, $this->lockValue, $this->ttl], 1 ); } public function unlock() { // 停止续期 if ($this->renewTimer) { swoole_timer_clear($this->renewTimer); } // 安全释放 $luaScript = " if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end "; $this->redis->eval( $luaScript, [$this->lockKey, $this->lockValue], 1 ); } }2.5 版本五：可重入锁同一线程多次获取同一把锁不应死锁：class ReentrantLock { private $redis; /** * 加锁（可重入） * 使用Hash结构：key=锁名, field=客户端ID, value=重入次数 */ public function lock($lockKey, $clientId, $ttl = 30) { $luaScript = " -- 锁不存在或当前客户端已持有 if redis.call('exists', KEYS[1]) == 0 or redis.call('hexists', KEYS[1], ARGV[1]) == 1 then -- 重入次数+1 redis.call('hincrby', KEYS[1], ARGV[1], 1) -- 设置过期时间 redis.call('expire', KEYS[1], ARGV[2]) return 1 else return 0 end "; return $this->redis->eval( $luaScript, [$lockKey, $clientId, $ttl], 1 ) === 1; } /** * 释放锁（可重入） * 重入次数-1，减到0时删除锁 */ public function unlock($lockKey, $clientId) { $luaScript = " -- 不是锁的持有者 if redis.call('hexists', KEYS[1], ARGV[1]) == 0 then return nil end -- 重入次数-1 local count = redis.call('hincrby', KEYS[1], ARGV[1], -1) if count > 0 then -- 还有重入次数，续期 redis.call('expire', KEYS[1], 30) return count else -- 重入次数归零，删除锁 redis.call('del', KEYS[1]) return 0 end "; return $this->redis->eval( $luaScript, [$lockKey, $clientId], 1 ); } } // 使用 $lock = new ReentrantLock(); $clientId = 'client-' . getmypid(); $lock->lock('lock:order', $clientId); // count=1 $lock->lock('lock:order', $clientId); // count=2（可重入） $lock->unlock('lock:order', $clientId); // count=1 $lock->unlock('lock:order', $clientId); // count=0, 锁删除三、Redlock 算法3.1 为什么需要 Redlock单 Redis 实例的分布式锁有一个问题：如果 Master 宕机，锁数据还没同步到 Slave，新 Master 上没有锁信息，其他客户端可以加锁成功，导致互斥性被破坏。Redlock 由 Redis 作者提出，通过多个独立 Redis 实例投票解决此问题。3.2 算法流程客户端 Redis集群 │ │── 加锁请求 ──→ Node1 ──→ 成功 (1/5) │── 加锁请求 ──→ Node2 ──→ 成功 (2/5) │── 加锁请求 ──→ Node3 ──→ 成功 (3/5) ← 超过半数 │── 加锁请求 ──→ Node4 ──→ 超时 │── 加锁请求 ──→ Node5 ──→ 超时 │ │ 加锁耗时 = 50ms │ 锁TTL = 30s │ 实际有效时间 = 30s - 50ms = 29.95s │ │←── 加锁成功（3/5节点同意）─────步骤：获取当前时间 T1依次向 N 个 Redis 节点请求加锁（相同的 key、value、较短TTL）获取当前时间 T2，计算加锁耗时 = T2 - T1如果加锁成功节点数 ≥ ⌈N/2⌉ + 1 且加锁耗时 < TTL，则加锁成功实际锁有效时间 = TTL - 加锁耗时如果加锁失败，向所有节点发送释放锁请求3.3 PHP 实现class Redlock { private $servers = []; public function __construct(array $redisServers) { foreach ($redisServers as $server) { $redis = new Redis(); $redis->connect($server['host'], $server['port']); if (isset($server['password'])) { $redis->auth($server['password']); } $this->servers[] = $redis; } } public function lock($resource, $ttl = 10000) { $lockValue = bin2hex(random_bytes(16)); $successCount = 0; $startTime = microtime(true) * 1000; // 向所有节点请求加锁 foreach ($this->servers as $redis) { try { $result = $redis->set( $resource, $lockValue, ['NX', 'PX' => $ttl] ); if ($result) { $successCount++; } } catch (Exception $e) { // 节点不可用，继续尝试其他节点 } } $elapsed = microtime(true) * 1000 - $startTime; $quorum = floor(count($this->servers) / 2) + 1; // 判断是否加锁成功 if ($successCount >= $quorum && $elapsed < $ttl) { // 返回锁信息（包含实际有效时间） return [ 'key' => $resource, 'value' => $lockValue, 'validity_time' => $ttl - intval($elapsed), ]; } // 加锁失败，释放所有已加的锁 $this->unlock($resource, $lockValue); return false; } public function unlock($resource, $lockValue) { $luaScript = " if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end "; foreach ($this->servers as $redis) { try { $redis->eval($luaScript, [$resource, $lockValue], 1); } catch (Exception $e) { // 忽略 } } } } // 使用 $redlock = new Redlock([ ['host' => '192.168.1.10', 'port' => 6379, 'password' => 'pass1'], ['host' => '192.168.1.11', 'port' => 6379, 'password' => 'pass2'], ['host' => '192.168.1.12', 'port' => 6379, 'password' => 'pass3'], ['host' => '192.168.1.13', 'port' => 6379, 'password' => 'pass4'], ['host' => '192.168.1.14', 'port' => 6379, 'password' => 'pass5'], ]); $lock = $redlock->lock('lock:order:1001', 10000); // TTL=10秒 if ($lock) { try { // 执行业务逻辑（注意在 validity_time 内完成） processOrder(); } finally { $redlock->unlock($lock['key'], $lock['value']); } } else { throw new Exception('获取分布式锁失败'); }3.4 Redlock 的争议Martin Kleppmann（《DDIA》作者）曾质疑 Redlock：时钟漂移可能导致锁失效GC 暂停可能导致客户端持有过期锁实际建议：如果对正确性要求极高（如金融场景），用 ZooKeeper 或 etcd如果是普通业务场景（如防重复下单），单 Redis + Lua 足够Redlock 适合需要比单 Redis 更高可靠性，但又不想引入 ZK 的场景四、实战：防重复下单class OrderService { private $redis; /** * 防重复下单 */ public function createOrder($userId, $productId, $quantity) { $lockKey = "lock:order:{$userId}:{$productId}"; $lockValue = uniqid('', true); // 加锁（10秒过期） $locked = $this->redis->set( $lockKey, $lockValue, ['NX', 'EX' => 10] ); if (!$locked) { throw new Exception('请勿重复提交，10秒后重试'); } try { // 1. 检查库存 $stock = $this->redis->get("stock:{$productId}"); if ($stock < $quantity) { throw new Exception('库存不足'); } // 2. 扣减库存 $this->redis->decrBy("stock:{$productId}", $quantity); // 3. 创建订单 $orderId = $this->db->insert('orders', [ 'user_id' => $userId, 'product_id' => $productId, 'quantity' => $quantity, 'status' => 'pending', 'created_at' => date('Y-m-d H:i:s'), ]); // 4. 发送延迟消息（30分钟未支付自动取消） $this->mq->publish('order_timeout', [ 'order_id' => $orderId, ], 1800); return $orderId; } catch (Exception $e) { // 回滚库存 $this->redis->incrBy("stock:{$productId}", $quantity); throw $e; } finally { // 安全释放锁 $lua = "if redis.call('get',KEYS[1])==ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end"; $this->redis->eval($lua, [$lockKey, $lockValue], 1); } } }五、各方案对比方案互斥性防死锁容错性性能复杂度SETNX+EXPIRE✅❌❌高低SET NX EX✅✅❌高低SET NX EX + Lua释放✅✅❌高中锁续期✅✅❌高中高可重入锁✅✅❌高高Redlock✅✅✅中高总结Redis 分布式锁的演进路线：SET NX EX 原子加锁 → 基础方案Lua 脚本安全释放 → 必须做后台续期 → 长任务必须做可重入 → 递归/嵌套场景需要Redlock → 对容错性要求高时考虑生产推荐：单 Redis + SET NX EX + Lua 释放 + 续期，覆盖 95% 场景。
前言MySQL 慢查询是影响应用性能的常见瓶颈。本文从慢查询日志分析入手，结合 EXPLAIN 执行计划解读，系统讲解索引优化和 SQL 调优方法。一、慢查询日志1.1 开启慢查询日志-- 查看慢查询配置 SHOW VARIABLES LIKE "%slow_query%"; -- 开启慢查询日志 SET GLOBAL slow_query_log = ON; SET GLOBAL long_query_time = 1; -- 超过 1 秒的查询 SET GLOBAL slow_query_log_file = "/var/log/mysql/slow.log"; -- 记录未使用索引的查询 SET GLOBAL log_queries_not_using_indexes = ON;1.2 使用 mysqldumpslow 分析# 按平均查询时间排序，取前 10 mysqldumpslow -s t -t 10 /var/log/mysql/slow.log # 按总耗时排序 mysqldumpslow -s at -t 10 /var/log/mysql/slow.log # 按次数排序 mysqldumpslow -s c -t 10 /var/log/mysql/slow.log二、EXPLAIN 执行计划2.1 基本用法EXPLAIN SELECT * FROM orders WHERE user_id = 10086 AND status = "paid";2.2 关键字段解读字段说明重点关注type访问类型至少达到 range 级别，避免 ALLkey实际使用的索引不为 NULLrows预估扫描行数越小越好Extra附加信息避免 Using filesort 和 Using temporary2.3 type 类型（从好到差）system > const > eq_ref > ref > range > index > ALLconst：主键或唯一索引等值查询eq_ref：JOIN 时使用主键或唯一索引ref：非唯一索引等值查询range：索引范围查询（BETWEEN、>、<、IN）index：扫描整个索引树ALL：全表扫描，必须优化三、索引优化3.1 创建合适的索引-- 单列索引 CREATE INDEX idx_user_id ON orders(user_id); -- 联合索引（注意最左前缀原则） CREATE INDEX idx_user_status ON orders(user_id, status, created_at); -- 覆盖索引（查询字段都在索引中） CREATE INDEX idx_cover ON orders(user_id, status, amount);3.2 最左前缀原则-- 联合索引 (a, b, c) -- 能命中索引： SELECT * FROM t WHERE a = 1; SELECT * FROM t WHERE a = 1 AND b = 2; SELECT * FROM t WHERE a = 1 AND b = 2 AND c = 3; -- 不能命中索引： SELECT * FROM t WHERE b = 2; SELECT * FROM t WHERE c = 3; SELECT * FROM t WHERE b = 2 AND c = 3;3.3 避免索引失效-- 函数操作导致索引失效 SELECT * FROM orders WHERE YEAR(created_at) = 2026; -- 优化： SELECT * FROM orders WHERE created_at >= "2026-01-01" AND created_at < "2027-01-01"; -- 隐式类型转换 SELECT * FROM orders WHERE order_no = 10086; -- order_no 是 varchar -- 优化： SELECT * FROM orders WHERE order_no = "10086"; -- LIKE 前导通配符 SELECT * FROM users WHERE name LIKE "%张"; -- 优化：使用全文索引或调整业务逻辑 SELECT * FROM users WHERE name LIKE "张%"; -- OR 导致索引失效 SELECT * FROM orders WHERE user_id = 1 OR status = "paid"; -- 优化：使用 UNION ALL SELECT * FROM orders WHERE user_id = 1 UNION ALL SELECT * FROM orders WHERE status = "paid" AND user_id != 1;四、SQL 调优实战4.1 分页优化-- 慢：OFFSET 大时性能差 SELECT * FROM orders ORDER BY id LIMIT 1000000, 20; -- 优化方案 1：延迟关联 SELECT * FROM orders o INNER JOIN (SELECT id FROM orders ORDER BY id LIMIT 1000000, 20) t ON o.id = t.id; -- 优化方案 2：游标分页（记住上一页最后一条 ID） SELECT * FROM orders WHERE id > 1000000 ORDER BY id LIMIT 20;4.2 批量插入-- 慢：循环单条插入 INSERT INTO logs (msg) VALUES ("log1"); INSERT INTO logs (msg) VALUES ("log2"); -- 优化：批量插入 INSERT INTO logs (msg) VALUES ("log1"), ("log2"), ("log3");4.3 使用 Redis 缓存热点数据import redis import json r = redis.Redis(host="localhost", port=6379, db=0) def get_user(user_id): # 先查缓存 cache_key = f"user:{user_id}" cached = r.get(cache_key) if cached: return json.loads(cached) # 查数据库 user = db.query("SELECT * FROM users WHERE id = %s", user_id) if user: r.setex(cache_key, 3600, json.dumps(user)) # 缓存 1 小时 return user五、总结MySQL 慢查询优化的核心思路：先通过慢查询日志找到问题 SQL，再用 EXPLAIN 分析执行计划，最后通过添加索引、调整 SQL 写法、引入缓存等方式优化。记住：不是所有慢查询都需要加索引，有时调整业务逻辑比优化 SQL 更有效。