分布式锁实现方案详解
分布式锁是分布式系统中的核心组件，用于在分布式场景下实现临界资源的互斥访问。本章将详细介绍分布式锁的实现原理、技术方案和最佳实践。
分布式锁的重要性
- 资源保护：防止多个进程同时访问共享资源
- 数据一致性：确保分布式环境下的数据操作原子性
- 业务安全：避免重复处理、超卖等业务问题
- 系统稳定：防止并发操作导致的系统异常
1. 分布式锁的基本要求
1.1 核心特性
分布式锁需要满足以下基本要求：
| 特性 | 描述 | 重要性 |
|---|---|---|
| 互斥性 | 同一时间只能有一个客户端持有锁 | 核心要求 |
| 防死锁 | 锁必须能够自动释放，避免死锁 | 系统稳定性 |
| 可重入性 | 同一客户端可以多次获取同一把锁 | 业务便利性 |
| 高性能 | 获取和释放锁的操作要快速 | 系统性能 |
| 高可用 | 锁服务本身要具备高可用性 | 系统可靠性 |
1.2 实现挑战
2. Redis分布式锁实现
2.1 基础实现
Redis分布式锁是最常用的分布式锁实现方案，具有高性能、易实现的特点。
获取锁
Redis分布式锁基础实现
java
1@Service2public class RedisDistributedLock {3 4 private final RedisTemplate<String, String> redisTemplate;5 private final String lockKey;6 private final String lockValue;7 private final long expireTime;8 9 public RedisDistributedLock(RedisTemplate<String, String> redisTemplate, 10 String lockKey, long expireTime) {11 this.redisTemplate = redisTemplate;12 this.lockKey = lockKey;13 this.lockValue = UUID.randomUUID().toString(); // 唯一标识14 this.expireTime = expireTime;15 }16 17 /**18 * 尝试获取锁19 * @param timeout 超时时间20 * @param unit 时间单位21 * @return 是否获取成功22 */23 public boolean tryLock(long timeout, TimeUnit unit) {24 long startTime = System.currentTimeMillis();25 long timeoutMillis = unit.toMillis(timeout);26 27 while (System.currentTimeMillis() - startTime < timeoutMillis) {28 // 使用SET NX EX命令原子性地设置锁29 Boolean success = redisTemplate.opsForValue()30 .setIfAbsent(lockKey, lockValue, expireTime, TimeUnit.MILLISECONDS);31 32 if (Boolean.TRUE.equals(success)) {33 return true; // 获取锁成功34 }35 36 // 获取锁失败，短暂等待后重试37 try {38 Thread.sleep(10);39 } catch (InterruptedException e) {40 Thread.currentThread().interrupt();41 return false;42 }43 }44 45 return false; // 超时未获取到锁46 }47 48 /**49 * 释放锁50 * @return 是否释放成功51 */52 public boolean releaseLock() {53 // 使用Lua脚本保证原子性54 String script = "if redis.call('get', KEYS[1]) == ARGV[1] then " +55 "return redis.call('del', KEYS[1]) " +56 "else return 0 end";57 58 Long result = redisTemplate.execute(59 new DefaultRedisScript<>(script, Long.class),60 Collections.singletonList(lockKey),61 lockValue62 );63 64 return Long.valueOf(1).equals(result);65 }66}使用示例
Redis分布式锁使用示例
java
1@Service2public class OrderService {3 4 private final RedisTemplate<String, String> redisTemplate;5 6 public void processOrder(String orderId) {7 String lockKey = "order:lock:" + orderId;8 RedisDistributedLock lock = new RedisDistributedLock(redisTemplate, lockKey, 30000);9 10 try {11 // 尝试获取锁，超时时间5秒12 if (lock.tryLock(5, TimeUnit.SECONDS)) {13 try {14 // 执行业务逻辑15 doProcessOrder(orderId);16 } finally {17 // 释放锁18 lock.releaseLock();19 }20 } else {21 throw new RuntimeException("Failed to acquire lock for order: " + orderId);22 }23 } catch (Exception e) {24 log.error("Error processing order: " + orderId, e);25 throw e;26 }27 }28 29 private void doProcessOrder(String orderId) {30 // 具体的订单处理逻辑31 log.info("Processing order: " + orderId);32 // ... 业务逻辑33 }34}2.2 看门狗机制
为了防止业务执行时间超过锁的过期时间，需要实现看门狗机制自动续期。
Redis分布式锁看门狗实现
java
1@Service2public class RedisDistributedLockWithWatchdog {3 4 private final RedisTemplate<String, String> redisTemplate;5 private final String lockKey;6 private final String lockValue;7 private final long expireTime;8 private final ScheduledExecutorService scheduler;9 private volatile boolean isLocked = false;10 private volatile ScheduledFuture<?> renewalTask;11 12 public RedisDistributedLockWithWatchdog(RedisTemplate<String, String> redisTemplate, 13 String lockKey, long expireTime) {14 this.redisTemplate = redisTemplate;15 this.lockKey = lockKey;16 this.lockValue = UUID.randomUUID().toString();17 this.expireTime = expireTime;18 this.scheduler = Executors.newScheduledThreadPool(1);19 }20 21 /**22 * 尝试获取锁（带看门狗）23 */24 public boolean tryLock(long timeout, TimeUnit unit) {25 long startTime = System.currentTimeMillis();26 long timeoutMillis = unit.toMillis(timeout);27 28 while (System.currentTimeMillis() - startTime < timeoutMillis) {29 Boolean success = redisTemplate.opsForValue()30 .setIfAbsent(lockKey, lockValue, expireTime, TimeUnit.MILLISECONDS);31 32 if (Boolean.TRUE.equals(success)) {33 isLocked = true;34 // 启动看门狗任务35 startWatchdog();36 return true;37 }38 39 try {40 Thread.sleep(10);41 } catch (InterruptedException e) {42 Thread.currentThread().interrupt();43 return false;44 }45 }46 47 return false;48 }49 50 /**51 * 启动看门狗任务52 */53 private void startWatchdog() {54 // 在过期时间的1/3时开始续期55 long renewalInterval = expireTime / 3;56 57 renewalTask = scheduler.scheduleAtFixedRate(() -> {58 if (isLocked) {59 try {60 // 检查锁是否仍然属于当前客户端61 String currentValue = redisTemplate.opsForValue().get(lockKey);62 if (lockValue.equals(currentValue)) {63 // 续期锁64 redisTemplate.expire(lockKey, expireTime, TimeUnit.MILLISECONDS);65 log.debug("Lock renewed for key: " + lockKey);66 } else {67 // 锁已被其他客户端获取，停止续期68 isLocked = false;69 if (renewalTask != null) {70 renewalTask.cancel(false);71 }72 }73 } catch (Exception e) {74 log.error("Error renewing lock: " + lockKey, e);75 isLocked = false;76 if (renewalTask != null) {77 renewalTask.cancel(false);78 }79 }80 }81 }, renewalInterval, renewalInterval, TimeUnit.MILLISECONDS);82 }83 84 /**85 * 释放锁86 */87 public boolean releaseLock() {88 if (!isLocked) {89 return false;90 }91 92 // 停止看门狗任务93 if (renewalTask != null) {94 renewalTask.cancel(false);95 }96 97 // 释放锁98 String script = "if redis.call('get', KEYS[1]) == ARGV[1] then " +99 "return redis.call('del', KEYS[1]) " +100 "else return 0 end";101 102 Long result = redisTemplate.execute(103 new DefaultRedisScript<>(script, Long.class),104 Collections.singletonList(lockKey),105 lockValue106 );107 108 boolean released = Long.valueOf(1).equals(result);109 if (released) {110 isLocked = false;111 }112 113 return released;114 }115 116 /**117 * 关闭资源118 */119 public void close() {120 if (renewalTask != null) {121 renewalTask.cancel(false);122 }123 scheduler.shutdown();124 }125}2.3 RedLock算法
RedLock是Redis官方推荐的分布式锁算法，通过多个独立的Redis节点来提高可靠性。
RedLock实现示例
java
1@Service2public class RedLockDistributedLock {3 4 private final List<RedisTemplate<String, String>> redisTemplates;5 private final String lockKey;6 private final String lockValue;7 private final long expireTime;8 private final int quorum;9 10 public RedLockDistributedLock(List<RedisTemplate<String, String>> redisTemplates, 11 String lockKey, long expireTime) {12 this.redisTemplates = redisTemplates;13 this.lockKey = lockKey;14 this.lockValue = UUID.randomUUID().toString();15 this.expireTime = expireTime;16 this.quorum = redisTemplates.size() / 2 + 1; // 多数派17 }18 19 /**20 * 尝试获取锁21 */22 public boolean tryLock(long timeout, TimeUnit unit) {23 long startTime = System.currentTimeMillis();24 long timeoutMillis = unit.toMillis(timeout);25 26 while (System.currentTimeMillis() - startTime < timeoutMillis) {27 long lockStartTime = System.currentTimeMillis();28 29 // 尝试在所有Redis节点上获取锁30 int successCount = 0;31 for (RedisTemplate<String, String> redisTemplate : redisTemplates) {32 try {33 Boolean success = redisTemplate.opsForValue()34 .setIfAbsent(lockKey, lockValue, expireTime, TimeUnit.MILLISECONDS);35 if (Boolean.TRUE.equals(success)) {36 successCount++;37 }38 } catch (Exception e) {39 log.warn("Failed to acquire lock on Redis node", e);40 }41 }42 43 // 检查是否获得多数派支持44 if (successCount >= quorum) {45 // 计算锁的有效时间46 long lockTime = System.currentTimeMillis() - lockStartTime;47 long validTime = expireTime - lockTime;48 49 if (validTime > 0) {50 return true; // 获取锁成功51 } else {52 // 锁的有效时间不足，释放所有锁53 releaseAllLocks();54 }55 } else {56 // 未获得多数派支持，释放已获取的锁57 releaseAllLocks();58 }59 60 // 短暂等待后重试61 try {62 Thread.sleep(10);63 } catch (InterruptedException e) {64 Thread.currentThread().interrupt();65 return false;66 }67 }68 69 return false;70 }71 72 /**73 * 释放所有锁74 */75 private void releaseAllLocks() {76 for (RedisTemplate<String, String> redisTemplate : redisTemplates) {77 try {78 String script = "if redis.call('get', KEYS[1]) == ARGV[1] then " +79 "return redis.call('del', KEYS[1]) " +80 "else return 0 end";81 82 redisTemplate.execute(83 new DefaultRedisScript<>(script, Long.class),84 Collections.singletonList(lockKey),85 lockValue86 );87 } catch (Exception e) {88 log.warn("Failed to release lock on Redis node", e);89 }90 }91 }92 93 /**94 * 释放锁95 */96 public boolean releaseLock() {97 return releaseAllLocks();98 }99}java
1// 获取锁2Boolean ok = redis.opsForValue().setIfAbsent(k, v, java.time.Duration.ofSeconds(30));3// 释放锁（Lua）4String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";5Long r = redis.execute(new org.springframework.data.redis.core.script.DefaultRedisScript<>(script, Long.class),6 java.util.Collections.singletonList(k), v);2.4 RedLock的优缺点
优点
- 高可靠性：通过多个独立Redis节点提高可用性
- 强一致性：需要多数派节点确认，保证锁的安全性
- 容错能力：部分节点故障不影响锁的正常工作
缺点
- 复杂性高：需要管理多个Redis节点
- 性能开销：需要与多个节点通信，增加延迟
- 时钟依赖：对时钟同步要求较高
- 资源消耗：需要更多的Redis实例
RedLock争议
RedLock算法在学术界存在争议，主要问题是时钟漂移可能导致锁的安全性无法保证。在实际使用中需要谨慎评估。
3. Zookeeper分布式锁实现
3.1 基本原理
Zookeeper分布式锁基于临时顺序节点实现，具有强一致性和自动释放的特点。
核心机制
- 临时节点：客户端断开连接时自动删除
- 顺序节点：保证获取锁的公平性
- 监听机制：监听前驱节点变化，实现阻塞等待
3.2 实现代码
Zookeeper分布式锁实现
java
1@Service2public class ZookeeperDistributedLock {3 4 private final CuratorFramework client;5 private final String lockPath;6 private final String lockName;7 private InterProcessMutex mutex;8 9 public ZookeeperDistributedLock(CuratorFramework client, String lockPath, String lockName) {10 this.client = client;11 this.lockPath = lockPath;12 this.lockName = lockName;13 this.mutex = new InterProcessMutex(client, lockPath + "/" + lockName);14 }15 16 /**17 * 尝试获取锁18 */19 public boolean tryLock(long timeout, TimeUnit unit) {20 try {21 return mutex.acquire(timeout, unit);22 } catch (Exception e) {23 log.error("Failed to acquire lock: " + lockName, e);24 return false;25 }26 }27 28 /**29 * 释放锁30 */31 public boolean releaseLock() {32 try {33 if (mutex.isAcquiredInThisProcess()) {34 mutex.release();35 return true;36 }37 return false;38 } catch (Exception e) {39 log.error("Failed to release lock: " + lockName, e);40 return false;41 }42 }43 44 /**45 * 检查是否持有锁46 */47 public boolean isLocked() {48 return mutex.isAcquiredInThisProcess();49 }50}3.3 手动实现Zookeeper锁
手动实现Zookeeper分布式锁
java
1@Service2public class ManualZookeeperLock {3 4 private final CuratorFramework client;5 private final String lockPath;6 private final String lockName;7 private String currentNode;8 private final CountDownLatch lockLatch = new CountDownLatch(1);9 10 public ManualZookeeperLock(CuratorFramework client, String lockPath, String lockName) {11 this.client = client;12 this.lockPath = lockPath;13 this.lockName = lockName;14 }15 16 /**17 * 尝试获取锁18 */19 public boolean tryLock(long timeout, TimeUnit unit) {20 try {21 // 1. 创建临时顺序节点22 currentNode = client.create()23 .withMode(CreateMode.EPHEMERAL_SEQUENTIAL)24 .forPath(lockPath + "/" + lockName + "-");25 26 // 2. 获取所有子节点27 List<String> children = client.getChildren().forPath(lockPath);28 children.sort(String::compareTo);29 30 // 3. 检查当前节点是否为最小序号31 String nodeName = currentNode.substring(currentNode.lastIndexOf('/') + 1);32 if (children.get(0).equals(nodeName)) {33 return true; // 获取锁成功34 }35 36 // 4. 监听前驱节点37 String previousNode = getPreviousNode(children, nodeName);38 if (previousNode != null) {39 // 设置监听器40 NodeCache nodeCache = new NodeCache(client, lockPath + "/" + previousNode);41 nodeCache.getListenable().addListener(() -> {42 if (nodeCache.getCurrentData() == null) {43 lockLatch.countDown(); // 前驱节点删除，通知等待线程44 }45 });46 nodeCache.start();47 48 // 等待前驱节点删除49 return lockLatch.await(timeout, unit);50 }51 52 return false;53 } catch (Exception e) {54 log.error("Failed to acquire lock: " + lockName, e);55 return false;56 }57 }58 59 /**60 * 获取前驱节点61 */62 private String getPreviousNode(List<String> children, String currentNode) {63 int index = children.indexOf(currentNode);64 if (index > 0) {65 return children.get(index - 1);66 }67 return null;68 }69 70 /**71 * 释放锁72 */73 public boolean releaseLock() {74 try {75 if (currentNode != null) {76 client.delete().forPath(currentNode);77 return true;78 }79 return false;80 } catch (Exception e) {81 log.error("Failed to release lock: " + lockName, e);82 return false;83 }84 }85}3.4 Zookeeper锁的特点
优点
- 强一致性：基于Zookeeper的强一致性保证
- 自动释放：客户端断开连接时自动释放锁
- 公平性：基于顺序节点，保证获取锁的公平性
- 可靠性：Zookeeper的高可用性保证锁服务的可靠性
缺点
- 性能较低：相比Redis锁，性能较低
- 复杂性：需要维护Zookeeper集群
- 网络开销：每次操作都需要网络通信
4. 数据库分布式锁实现
4.1 基于唯一索引的锁
数据库分布式锁实现
java
1@Service2public class DatabaseDistributedLock {3 4 private final JdbcTemplate jdbcTemplate;5 private final String lockKey;6 private final String owner;7 8 public DatabaseDistributedLock(JdbcTemplate jdbcTemplate, String lockKey, String owner) {9 this.jdbcTemplate = jdbcTemplate;10 this.lockKey = lockKey;11 this.owner = owner;12 }13 14 /**15 * 尝试获取锁16 */17 public boolean tryLock(long timeout, TimeUnit unit) {18 long startTime = System.currentTimeMillis();19 long timeoutMillis = unit.toMillis(timeout);20 21 while (System.currentTimeMillis() - startTime < timeoutMillis) {22 try {23 // 尝试插入锁记录24 String sql = "INSERT INTO distributed_lock (lock_key, owner, expire_at) VALUES (?, ?, ?)";25 long expireAt = System.currentTimeMillis() + timeoutMillis;26 27 jdbcTemplate.update(sql, lockKey, owner, expireAt);28 return true; // 插入成功，获取锁29 30 } catch (DuplicateKeyException e) {31 // 锁已存在，检查是否过期32 if (isLockExpired()) {33 // 删除过期锁并重试34 releaseExpiredLock();35 continue;36 }37 38 // 锁未过期，等待后重试39 try {40 Thread.sleep(100);41 } catch (InterruptedException ie) {42 Thread.currentThread().interrupt();43 return false;44 }45 }46 }47 48 return false; // 超时未获取到锁49 }50 51 /**52 * 检查锁是否过期53 */54 private boolean isLockExpired() {55 String sql = "SELECT expire_at FROM distributed_lock WHERE lock_key = ?";56 try {57 Long expireAt = jdbcTemplate.queryForObject(sql, Long.class, lockKey);58 return expireAt != null && expireAt < System.currentTimeMillis();59 } catch (EmptyResultDataAccessException e) {60 return true; // 锁不存在，视为过期61 }62 }63 64 /**65 * 释放过期锁66 */67 private void releaseExpiredLock() {68 String sql = "DELETE FROM distributed_lock WHERE lock_key = ? AND expire_at < ?";69 jdbcTemplate.update(sql, lockKey, System.currentTimeMillis());70 }71 72 /**73 * 释放锁74 */75 public boolean releaseLock() {76 String sql = "DELETE FROM distributed_lock WHERE lock_key = ? AND owner = ?";77 int rows = jdbcTemplate.update(sql, lockKey, owner);78 return rows > 0;79 }80}4.2 数据库锁表结构
分布式锁表结构
sql
1CREATE TABLE distributed_lock (2 id BIGINT AUTO_INCREMENT PRIMARY KEY,3 lock_key VARCHAR(128) NOT NULL UNIQUE,4 owner VARCHAR(64) NOT NULL,5 expire_at BIGINT NOT NULL,6 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,7 updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,8 INDEX idx_lock_key (lock_key),9 INDEX idx_expire_at (expire_at)10);4.3 数据库锁的特点
优点
- 实现简单：基于数据库，实现相对简单
- 强一致性：利用数据库的事务特性保证一致性
- 可靠性高：数据库的高可用性保证锁的可靠性
缺点
- 性能较低：数据库操作相对较慢
- 单点瓶颈：容易成为系统瓶颈
- 资源消耗：占用数据库连接和资源
5. 分布式锁选型对比
5.1 技术方案对比
| 特性 | Redis锁 | Zookeeper锁 | 数据库锁 |
|---|---|---|---|
| 性能 | 极高 | 中等 | 较低 |
| 一致性 | 最终一致 | 强一致 | 强一致 |
| 可靠性 | 中等 | 高 | 高 |
| 实现复杂度 | 简单 | 中等 | 简单 |
| 自动释放 | 需要TTL | 自动 | 需要TTL |
| 公平性 | 无 | 有 | 无 |
| 可重入 | 需要实现 | 支持 | 需要实现 |
5.2 选型建议
选择Redis锁的场景
- 高性能要求：对锁的性能要求极高
- 简单业务：业务逻辑相对简单，不需要强一致性
- 快速实现：需要快速实现分布式锁功能
- 大规模部署：需要支持大规模并发访问
选择Zookeeper锁的场景
- 强一致性要求：对数据一致性要求极高
- 复杂业务：业务逻辑复杂，需要可靠的锁机制
- 公平性要求：需要保证获取锁的公平性
- 自动释放：希望锁能够自动释放，避免死锁
选择数据库锁的场景
- 简单实现：希望用最简单的方式实现分布式锁
- 小规模应用：应用规模较小，性能要求不高
- 过渡方案：作为临时或过渡的分布式锁方案
- 已有数据库：系统中已经有数据库，不想引入额外组件
6. 分布式锁最佳实践
6.1 锁的设计原则
锁粒度设计
锁粒度设计示例
java
1public class LockGranularityExample {2 3 // 粗粒度锁：整个用户级别的锁4 public void processUserData(String userId) {5 String lockKey = "user:lock:" + userId;6 // 使用分布式锁保护整个用户数据处理过程7 }8 9 // 细粒度锁：具体操作级别的锁10 public void updateUserBalance(String userId, String operation) {11 String lockKey = "user:balance:lock:" + userId + ":" + operation;12 // 只锁定具体的余额操作13 }14 15 // 分层锁：多级锁保护16 public void complexOperation(String userId, String resourceId) {17 // 先获取用户锁18 String userLockKey = "user:lock:" + userId;19 // 再获取资源锁20 String resourceLockKey = "resource:lock:" + resourceId;21 // 避免死锁：按固定顺序获取锁22 }23}锁超时设置
锁超时设置示例
java
1public class LockTimeoutExample {2 3 // 根据业务复杂度设置超时时间4 public void processSimpleTask() {5 // 简单任务：30秒超时6 long timeout = 30_000;7 }8 9 public void processComplexTask() {10 // 复杂任务：5分钟超时11 long timeout = 5 * 60 * 1000;12 }13 14 public void processBatchTask() {15 // 批量任务：30分钟超时16 long timeout = 30 * 60 * 1000;17 }18 19 // 动态超时设置20 public long calculateTimeout(int dataSize, int complexity) {21 long baseTimeout = 10_000; // 基础超时时间22 long dataTimeout = dataSize * 100; // 数据量相关超时23 long complexityTimeout = complexity * 5_000; // 复杂度相关超时24 return baseTimeout + dataTimeout + complexityTimeout;25 }26}6.2 常见问题与解决方案
误删锁问题
防止误删锁示例
java
1public class SafeLockRelease {2 3 public boolean safeReleaseLock(String lockKey, String lockValue) {4 // 使用Lua脚本保证原子性检查和删除5 String script = "if redis.call('get', KEYS[1]) == ARGV[1] then " +6 "return redis.call('del', KEYS[1]) " +7 "else return 0 end";8 9 Long result = redisTemplate.execute(10 new DefaultRedisScript<>(script, Long.class),11 Collections.singletonList(lockKey),12 lockValue13 );14 15 return Long.valueOf(1).equals(result);16 }17}可重入锁实现
可重入锁实现示例
java
1public class ReentrantDistributedLock {2 3 private final RedisTemplate<String, String> redisTemplate;4 private final String lockKey;5 private final String lockValue;6 private final ThreadLocal<Integer> lockCount = new ThreadLocal<>();7 8 public boolean tryLock(long timeout, TimeUnit unit) {9 Integer count = lockCount.get();10 if (count != null && count > 0) {11 // 当前线程已持有锁，增加计数12 lockCount.set(count + 1);13 return true;14 }15 16 // 尝试获取锁17 Boolean success = redisTemplate.opsForValue()18 .setIfAbsent(lockKey, lockValue + ":1", timeout, unit);19 20 if (Boolean.TRUE.equals(success)) {21 lockCount.set(1);22 return true;23 }24 25 return false;26 }27 28 public boolean releaseLock() {29 Integer count = lockCount.get();30 if (count == null || count <= 0) {31 return false;32 }33 34 if (count == 1) {35 // 最后一次释放，删除锁36 lockCount.remove();37 return safeReleaseLock(lockKey, lockValue + ":1");38 } else {39 // 减少计数40 lockCount.set(count - 1);41 return true;42 }43 }44}读写锁实现
读写锁实现示例
java
1public class ReadWriteDistributedLock {2 3 private final RedisTemplate<String, String> redisTemplate;4 private final String lockKey;5 private final String clientId;6 7 public boolean tryReadLock(long timeout, TimeUnit unit) {8 String readLockKey = lockKey + ":read";9 String writeLockKey = lockKey + ":write";10 11 // 检查是否有写锁12 String writeLock = redisTemplate.opsForValue().get(writeLockKey);13 if (writeLock != null && !writeLock.equals(clientId)) {14 return false; // 有写锁且不是当前客户端15 }16 17 // 尝试获取读锁18 return redisTemplate.opsForValue()19 .setIfAbsent(readLockKey + ":" + clientId, "1", timeout, unit);20 }21 22 public boolean tryWriteLock(long timeout, TimeUnit unit) {23 String readLockKey = lockKey + ":read";24 String writeLockKey = lockKey + ":write";25 26 // 检查是否有读锁27 Set<String> readLocks = redisTemplate.keys(readLockKey + ":*");28 if (!readLocks.isEmpty()) {29 return false; // 有读锁存在30 }31 32 // 尝试获取写锁33 return redisTemplate.opsForValue()34 .setIfAbsent(writeLockKey, clientId, timeout, unit);35 }36}6.3 监控与告警
锁监控指标
锁监控示例
java
1@Component2public class LockMonitorService {3 4 private final MeterRegistry meterRegistry;5 6 public void recordLockAcquisition(String lockKey, long duration, boolean success) {7 // 记录锁获取次数8 Counter.builder("distributed_lock.acquisitions")9 .tag("lock_key", lockKey)10 .tag("result", success ? "success" : "failure")11 .register(meterRegistry)12 .increment();13 14 // 记录锁获取时间15 Timer.builder("distributed_lock.acquisition_time")16 .tag("lock_key", lockKey)17 .register(meterRegistry)18 .record(duration, TimeUnit.MILLISECONDS);19 }20 21 public void recordLockHoldingTime(String lockKey, long duration) {22 // 记录锁持有时间23 Timer.builder("distributed_lock.holding_time")24 .tag("lock_key", lockKey)25 .register(meterRegistry)26 .record(duration, TimeUnit.MILLISECONDS);27 }28 29 @Scheduled(fixedRate = 60000) // 每分钟检查一次30 public void checkLockHealth() {31 // 检查锁的健康状态32 // 例如：检查是否有长时间未释放的锁33 // 检查锁的获取失败率等34 }35}7. 分布式锁面试题精选
7.1 基础概念
Q1: 什么是分布式锁？为什么需要分布式锁？
答: 分布式锁是分布式系统中的一种同步机制，用于在分布式环境下实现临界资源的互斥访问。
为什么需要分布式锁:
- 资源保护：防止多个进程同时访问共享资源
- 数据一致性：确保分布式环境下的数据操作原子性
- 业务安全：避免重复处理、超卖等业务问题
- 系统稳定：防止并发操作导致的系统异常
Q2: 分布式锁需要满足哪些基本要求？
答: 分布式锁需要满足以下基本要求：
- 互斥性：同一时间只能有一个客户端持有锁
- 防死锁：锁必须能够自动释放，避免死锁
- 可重入性：同一客户端可以多次获取同一把锁
- 高性能：获取和释放锁的操作要快速
- 高可用：锁服务本身要具备高可用性
7.2 Redis分布式锁
Q3: Redis分布式锁的实现原理是什么？
答: Redis分布式锁基于Redis的SET命令的原子性实现：
-
获取锁：使用
SET key value NX EX ttl命令- NX：只有当key不存在时才设置
- EX：设置过期时间
- 返回OK表示获取成功，返回nil表示获取失败
-
释放锁：使用Lua脚本保证原子性
- 先检查锁是否属于当前客户端
- 如果是，则删除锁；否则不删除
Redis分布式锁核心实现
java
1// 获取锁2Boolean success = redisTemplate.opsForValue()3 .setIfAbsent(lockKey, lockValue, expireTime, TimeUnit.MILLISECONDS);45// 释放锁（Lua脚本）6String script = "if redis.call('get', KEYS[1]) == ARGV[1] then " +7 "return redis.call('del', KEYS[1]) " +8 "else return 0 end";9Long result = redisTemplate.execute(script, lockKey, lockValue);Q4: Redis分布式锁存在哪些问题？如何解决？
答: Redis分布式锁存在以下问题：
-
锁误删问题：
- 问题：客户端A获取锁后，业务执行时间超过锁过期时间，锁被自动释放，客户端B获取锁，然后客户端A释放锁时误删了客户端B的锁
- 解决：在释放锁时检查锁是否属于当前客户端
-
锁续期问题：
- 问题：业务执行时间可能超过锁的过期时间
- 解决：实现看门狗机制，定期续期锁
-
单点故障问题：
- 问题：Redis单点故障导致锁服务不可用
- 解决：使用Redis集群或RedLock算法
看门狗机制示例
java
1private void startWatchdog() {2 long renewalInterval = expireTime / 3;3 4 renewalTask = scheduler.scheduleAtFixedRate(() -> {5 if (isLocked) {6 String currentValue = redisTemplate.opsForValue().get(lockKey);7 if (lockValue.equals(currentValue)) {8 // 续期锁9 redisTemplate.expire(lockKey, expireTime, TimeUnit.MILLISECONDS);10 }11 }12 }, renewalInterval, renewalInterval, TimeUnit.MILLISECONDS);13}7.3 Zookeeper分布式锁
Q5: Zookeeper分布式锁的实现原理是什么？
答: Zookeeper分布式锁基于临时顺序节点实现：
- 创建节点：客户端在锁目录下创建临时顺序节点
- 检查序号：检查当前节点是否为最小序号
- 监听前驱：如果不是最小序号，监听前驱节点
- 获取锁：前驱节点删除后，重新检查序号
Zookeeper锁核心逻辑
java
1// 创建临时顺序节点2String currentNode = client.create()3 .withMode(CreateMode.EPHEMERAL_SEQUENTIAL)4 .forPath(lockPath + "/" + lockName + "-");56// 获取所有子节点并排序7List<String> children = client.getChildren().forPath(lockPath);8children.sort(String::compareTo);910// 检查是否为最小序号11String nodeName = currentNode.substring(currentNode.lastIndexOf('/') + 1);12if (children.get(0).equals(nodeName)) {13 return true; // 获取锁成功14}1516// 监听前驱节点17String previousNode = getPreviousNode(children, nodeName);18// 设置监听器等待前驱节点删除Q6: Zookeeper分布式锁相比Redis锁有什么优势？
答: Zookeeper分布式锁相比Redis锁有以下优势：
- 强一致性：基于Zookeeper的强一致性保证
- 自动释放：客户端断开连接时自动释放锁
- 公平性：基于顺序节点，保证获取锁的公平性
- 可靠性：Zookeeper的高可用性保证锁服务的可靠性
缺点：
- 性能较低：相比Redis锁，性能较低
- 复杂性：需要维护Zookeeper集群
- 网络开销：每次操作都需要网络通信
7.4 高级问题
Q7: 如何实现可重入分布式锁？
答: 可重入分布式锁需要记录锁的持有次数：
可重入锁实现
java
1public class ReentrantDistributedLock {2 private final ThreadLocal<Integer> lockCount = new ThreadLocal<>();3 4 public boolean tryLock(long timeout, TimeUnit unit) {5 Integer count = lockCount.get();6 if (count != null && count > 0) {7 // 当前线程已持有锁，增加计数8 lockCount.set(count + 1);9 return true;10 }11 12 // 尝试获取锁13 Boolean success = redisTemplate.opsForValue()14 .setIfAbsent(lockKey, lockValue + ":1", timeout, unit);15 16 if (Boolean.TRUE.equals(success)) {17 lockCount.set(1);18 return true;19 }20 21 return false;22 }23 24 public boolean releaseLock() {25 Integer count = lockCount.get();26 if (count == null || count <= 0) {27 return false;28 }29 30 if (count == 1) {31 // 最后一次释放，删除锁32 lockCount.remove();33 return safeReleaseLock(lockKey, lockValue + ":1");34 } else {35 // 减少计数36 lockCount.set(count - 1);37 return true;38 }39 }40}Q8: 如何设计分布式锁的监控系统？
答: 分布式锁监控系统需要关注以下指标：
-
性能指标：
- 锁获取成功率
- 锁获取平均时间
- 锁持有时间分布
-
业务指标：
- 锁竞争情况
- 锁等待队列长度
- 锁超时情况
-
系统指标：
- 锁服务可用性
- 锁服务响应时间
- 锁服务错误率
锁监控实现
java
1@Component2public class LockMonitorService {3 4 public void recordLockAcquisition(String lockKey, long duration, boolean success) {5 // 记录锁获取次数6 Counter.builder("distributed_lock.acquisitions")7 .tag("lock_key", lockKey)8 .tag("result", success ? "success" : "failure")9 .register(meterRegistry)10 .increment();11 12 // 记录锁获取时间13 Timer.builder("distributed_lock.acquisition_time")14 .tag("lock_key", lockKey)15 .register(meterRegistry)16 .record(duration, TimeUnit.MILLISECONDS);17 }18 19 @Scheduled(fixedRate = 60000)20 public void checkLockHealth() {21 // 检查长时间未释放的锁22 // 检查锁的获取失败率23 // 发送告警24 }25}8. 总结
分布式锁是分布式系统中的重要组件，不同的实现方案各有优缺点。在实际应用中，需要根据业务场景、性能要求和一致性要求选择合适的分布式锁方案。
关键要点
- 选择合适的锁方案：根据业务需求选择Redis、Zookeeper或数据库锁
- 解决锁的问题：正确处理锁的误删、续期、可重入等问题
- 保证锁的安全性：使用Lua脚本、看门狗机制等技术保证锁的安全性
- 监控和优化：建立完善的监控体系，持续优化锁的性能
通过深入理解和熟练运用这些技术，我们能够构建出安全、高效的分布式锁系统。
评论