Redis 分布式锁实现：从 SET NX 到 Redlock 算法 前言分布式锁是分布式系统中控制资源访问的核心机制。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 到索引调优 前言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 更有效。