前期， 我们介绍了什么是分布式锁及分布式锁应用场景， 今天我们基于Redis方案来实现分布式锁的应用。
1. 基于Redis分布式锁方案介绍
基于Redis实现的分布式锁是分布式系统中控制资源访问的常用方案，利用Redis的原子操作和高性能特性实现跨进程的锁机制。我们需根据业务特性合理设计，避免引入新的问题（如死锁、锁失效）。
基于Redis方案的分布式具有以下特点：
- 可重入性：同一线程可多次获取同一把锁，通过 ThreadLocal 计数。
- 锁续期（看门狗机制）：自动延长锁的过期时间，防止业务未执行完锁就过期。
- 公平 / 非公平锁：支持通过构造函数指定是否公平锁。
- 异常处理：包含完整的异常处理和资源释放逻辑。
- 原子操作：能够使用 Lua 脚本保证解锁的原子性。
1.1 核心原理
-
原子性加锁
通过Redis的原子命令SET key value NX PX timeout实现：NX（Not eXists）：仅当key不存在时设置值，保证互斥性。PX timeout：设置锁的过期时间（毫秒），避免死锁（如持有锁的进程崩溃时自动释放）。
-
唯一标识
锁的value使用唯一ID（如UUID），确保锁只能被持有者释放，防止误删。 -
原子性解锁
使用Lua脚本保证解锁的原子性：if redis.call("GET", KEYS[1]) == ARGV[1] thenreturn redis.call("DEL", KEYS[1]) elsereturn 0 end先验证锁的持有者，再删除锁，避免误释放其他进程持有的锁。
1.2 优缺点
-
优点
- 高性能：Redis基于内存操作，加锁和解锁延迟极低。
- 高可用：通过Redis集群（如Sentinel、Cluster）避免单点故障。
- 灵活配置：支持设置锁的过期时间、重试策略等参数。
-
缺点
- 主从复制延迟：主从架构中，主节点写入锁后未同步到从节点就崩溃，可能导致锁丢失（Redlock算法可部分解决）。
- 过期时间难控制：若业务执行时间超过锁的过期时间，可能导致多个进程同时持有锁。
1.3 典型实现方式
-
简单实现（原生Redis命令）
直接使用Redis客户端（如Jedis、Lettuce）调用SET和Lua脚本，适合轻量级场景。 -
Redisson框架
提供分布式锁的高级抽象（如可重入锁、公平锁、读写锁），内置看门狗机制自动续期锁：// Redisson可重入锁示例 RLock lock = redissonClient.getLock("myLock"); lock.lock(); // 自动续期，默认30秒 try {// 业务逻辑 } finally {lock.unlock(); } -
Redlock算法
针对主从复制缺陷，在多个独立的Redis节点上获取锁，多数节点成功时才认为加锁成功，提升可靠性，但牺牲部分性能。
1.4 关键参数配置
- 锁过期时间：需根据业务执行时间合理设置，避免过短导致锁提前释放，或过长导致资源长时间被占用。
- 重试策略：获取锁失败时的重试次数和间隔，避免频繁重试耗尽资源。
- 续期机制：通过看门狗自动延长锁的过期时间，确保业务执行期间锁不会过期。
1.5 适用场景
- 高并发场景：如秒杀、库存扣减，利用Redis高性能快速响应。
- 异步任务：如定时任务去重执行，通过锁避免多个节点重复处理。
- 缓存重建：防止缓存失效时多个请求同时重建缓存，造成缓存击穿。
1.6 注意事项
- 避免锁粒度过大：只在关键操作上加锁，减少锁持有时间。
- 异常处理：使用
try-finally确保锁最终被释放。 - 监控与告警：监控锁的持有时间、竞争情况，及时发现异常。
2. 实现代码
以下是基于Redis方案的分布式锁的重要实现代码片段（仅供参考）。
使用时需要在Maven文件中添加 Jedis 组件依赖：
<dependency><groupId>redis.clients</groupId><artifactId>jedis</artifactId><version>4.4.3</version>
</dependency>
RedisDistributedLock.java
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.params.SetParams;import java.util.Collections;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;/*** Redis分布式锁实现，基于Redis的SETNX和Lua脚本机制* 支持可重入、锁续期（看门狗机制）和公平锁特性*/
public class RedisDistributedLock implements Lock {private static final String LOCK_SUCCESS = "OK";private static final Long RELEASE_SUCCESS = 1L;private static final String SET_IF_NOT_EXIST = "NX";private static final String SET_WITH_EXPIRE_TIME = "PX";// 锁续期的间隔时间（毫秒），默认是锁超时时间的1/3private static final long RENEWAL_INTERVAL_RATIO = 3;private final JedisPool jedisPool;private final String lockKey;private final String clientId; // 客户端唯一标识private final long expireTime; // 锁超时时间（毫秒）private final boolean isFair; // 是否公平锁// 可重入计数private final ThreadLocal<AtomicInteger> reentrantCount = ThreadLocal.withInitial(() -> new AtomicInteger(0));// 看门狗线程private ThreadLocal<WatchDog> watchDog = new ThreadLocal<>();public RedisDistributedLock(JedisPool jedisPool, String lockKey, long expireTime, boolean isFair) {this.jedisPool = jedisPool;this.lockKey = lockKey;this.clientId = generateClientId();this.expireTime = expireTime;this.isFair = isFair;}@Overridepublic void lock() {if (!tryLock()) {// 等待并重试waitAndRetry();}}@Overridepublic void lockInterruptibly() throws InterruptedException {if (Thread.interrupted()) {throw new InterruptedException();}if (!tryLock()) {waitAndHandleInterrupt();}}@Overridepublic boolean tryLock() {// 检查是否已持有锁（可重入）if (isHeldByCurrentThread()) {reentrantCount.get().incrementAndGet();return true;}try (Jedis jedis = jedisPool.getResource()) {// SET key value NX PX expireTimeString result = jedis.set(lockKey, clientId, new SetParams().nx().px(expireTime));if (LOCK_SUCCESS.equals(result)) {reentrantCount.get().set(1);startWatchDog(); // 启动看门狗线程return true;}return false;}}@Overridepublic boolean tryLock(long time, TimeUnit unit) throws InterruptedException {long startTime = System.currentTimeMillis();long timeoutMillis = unit.toMillis(time);if (tryLock()) {return true;}while (System.currentTimeMillis() - startTime < timeoutMillis) {if (Thread.interrupted()) {throw new InterruptedException();}if (tryLock()) {return true;}Thread.sleep(100); // 避免CPU空转}return false;}@Overridepublic void unlock() {if (!isHeldByCurrentThread()) {throw new IllegalMonitorStateException("Attempt to unlock lock, not locked by current thread");}// 可重入锁计数减1int count = reentrantCount.get().decrementAndGet();if (count > 0) {return; // 仍持有锁，不释放}try {// 释放锁前停止看门狗stopWatchDog();// 使用Lua脚本保证原子性释放锁try (Jedis jedis = jedisPool.getResource()) {String script = "if redis.call('get', KEYS[1]) == ARGV[1] then " +"return redis.call('del', KEYS[1]) " +"else " +"return 0 " +"end";Object result = jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(clientId));if (!RELEASE_SUCCESS.equals(result)) {throw new IllegalMonitorStateException("Failed to release lock");}}} finally {// 清理ThreadLocalreentrantCount.remove();watchDog.remove();}}@Overridepublic Condition newCondition() {throw new UnsupportedOperationException("Conditions are not supported by this lock");}// 生成客户端唯一标识private String generateClientId() {return Thread.currentThread().getName() + "-" + System.currentTimeMillis() + "-" + (int)(Math.random() * 10000);}// 判断当前线程是否持有锁private boolean isHeldByCurrentThread() {AtomicInteger count = reentrantCount.get();return count.get() > 0;}// 等待并重试获取锁（非公平）private void waitAndRetry() {while (true) {if (tryLock()) {return;}try {Thread.sleep(100); // 避免CPU空转} catch (InterruptedException e) {Thread.currentThread().interrupt();return;}}}// 等待并处理中断private void waitAndHandleInterrupt() throws InterruptedException {while (true) {if (Thread.interrupted()) {throw new InterruptedException();}if (tryLock()) {return;}Thread.sleep(100); // 避免CPU空转}}// 启动看门狗线程，自动续期锁private void startWatchDog() {WatchDog dog = new WatchDog(expireTime / RENEWAL_INTERVAL_RATIO);watchDog.set(dog);dog.start();}// 停止看门狗线程private void stopWatchDog() {WatchDog dog = watchDog.get();if (dog != null) {dog.interrupt();}}/*** 看门狗线程，负责自动续期锁*/private class WatchDog extends Thread {private final long renewalInterval;private boolean running = true;public WatchDog(long renewalInterval) {this.renewalInterval = renewalInterval;setDaemon(true);setName("LockWatchDog-" + lockKey);}@Overridepublic void run() {while (running && !isInterrupted()) {try {Thread.sleep(renewalInterval);// 续期锁try (Jedis jedis = jedisPool.getResource()) {String script = "if redis.call('get', KEYS[1]) == ARGV[1] then " +"return redis.call('pexpire', KEYS[1], ARGV[2]) " +"else " +"return 0 " +"end";jedis.eval(script, Collections.singletonList(lockKey), List.of(clientId, String.valueOf(expireTime)));}} catch (InterruptedException e) {running = false;Thread.currentThread().interrupt();} catch (Exception e) {// 记录异常但继续运行e.printStackTrace();}}}}
}
RedisLockExample.java
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;/*** Redis分布式锁使用示例*/
public class RedisLockExample {private static final String REDIS_HOST = "localhost";private static final int REDIS_PORT = 6379;private static final String LOCK_KEY = "distributed:lock:example";private static final long LOCK_EXPIRE_TIME = 30000; // 30秒public static void main(String[] args) {// 创建Redis连接池JedisPoolConfig poolConfig = new JedisPoolConfig();poolConfig.setMaxTotal(100);poolConfig.setMaxIdle(20);poolConfig.setMinIdle(5);poolConfig.setTestOnBorrow(true);JedisPool jedisPool = new JedisPool(poolConfig, REDIS_HOST, REDIS_PORT);// 创建分布式锁实例RedisDistributedLock lock = new RedisDistributedLock(jedisPool, LOCK_KEY, LOCK_EXPIRE_TIME, false);// 模拟多线程竞争ExecutorService executor = Executors.newFixedThreadPool(5);for (int i = 0; i < 10; i++) {executor.submit(() -> {try {// 获取锁（带超时）if (lock.tryLock(5, TimeUnit.SECONDS)) {try {System.out.println(Thread.currentThread().getName() + " 获取到锁");// 模拟业务操作Thread.sleep(2000);} finally {lock.unlock();System.out.println(Thread.currentThread().getName() + " 释放锁");}} else {System.out.println(Thread.currentThread().getName() + " 获取锁超时");}} catch (InterruptedException e) {Thread.currentThread().interrupt();}});}executor.shutdown();try {executor.awaitTermination(1, TimeUnit.MINUTES);} catch (InterruptedException e) {e.printStackTrace();}// 关闭Redis连接池jedisPool.close();}
}
：初识Git)
