Redis分布式锁的几种实现方法详解

分布式锁是解决分布式系统并发问题的关键组件，Redis因其高性能和丰富的数据结构成为实现分布式锁的常用选择。以下是Redis分布式锁的几种主流实现方式及其技术细节：

---

一、基础实现方案

1. SETNX + EXPIRE（基本版）

// 加锁
Boolean locked = redisTemplate.opsForValue().setIfAbsent("lock:order:123", "1", 30, TimeUnit.SECONDS);
// 解锁（直接删除）
redisTemplate.delete("lock:order:123");
// 加锁
Boolean locked = redisTemplate.opsForValue().setIfAbsent("lock:order:123", "1", 30, TimeUnit.SECONDS);

// 解锁（直接删除）
redisTemplate.delete("lock:order:123");
// 加锁
Boolean locked = redisTemplate.opsForValue().setIfAbsent("lock:order:123", "1", 30, TimeUnit.SECONDS);

// 解锁（直接删除）
redisTemplate.delete("lock:order:123");

问题：非原子性操作可能导致死锁（SETNX成功但EXPIRE失败）

2. SET扩展参数（推荐基础方案）

// 原子性加锁（Redis 2.6.12+）
Boolean locked = redisTemplate.opsForValue().setIfAbsent(
    "lock:order:123", 
    UUID.randomUUID().toString(),
    30, 
    TimeUnit.SECONDS
);
// 解锁（Lua脚本保证原子性）
String script = 
    "if redis.call('get', KEYS[1]) == ARGV[1] then " +
    "   return redis.call('del', KEYS[1]) " +
    "else " +
    "   return 0 " +
    "end";
redisTemplate.execute(
    new DefaultRedisScript<>(script, Long.class),
    Collections.singletonList("lock:order:123"),
    lockValue
);
// 原子性加锁（Redis 2.6.12+）
Boolean locked = redisTemplate.opsForValue().setIfAbsent(
    "lock:order:123", 
    UUID.randomUUID().toString(),
    30, 
    TimeUnit.SECONDS
);

// 解锁（Lua脚本保证原子性）
String script = 
    "if redis.call('get', KEYS[1]) == ARGV[1] then " +
    "   return redis.call('del', KEYS[1]) " +
    "else " +
    "   return 0 " +
    "end";
redisTemplate.execute(
    new DefaultRedisScript<>(script, Long.class),
    Collections.singletonList("lock:order:123"),
    lockValue
);
// 原子性加锁（Redis 2.6.12+）
Boolean locked = redisTemplate.opsForValue().setIfAbsent(
    "lock:order:123", 
    UUID.randomUUID().toString(),
    30, 
    TimeUnit.SECONDS
);

// 解锁（Lua脚本保证原子性）
String script = 
    "if redis.call('get', KEYS[1]) == ARGV[1] then " +
    "   return redis.call('del', KEYS[1]) " +
    "else " +
    "   return 0 " +
    "end";
redisTemplate.execute(
    new DefaultRedisScript<>(script, Long.class),
    Collections.singletonList("lock:order:123"),
    lockValue
);

优点：原子性操作，避免误删其他客户端锁

---

二、高可用方案

3. RedLock算法（Redis官方推荐）

// 需要至少3个独立的Redis实例
List<RedisTemplate> redisTemplates = Arrays.asList(redis1, redis2, redis3);
// 获取锁
String lockKey = "lock:order:123";
String lockValue = UUID.randomUUID().toString();
int lockExpireTime = 30; // 秒
int successCount = 0;
long startTime = System.currentTimeMillis();
for (RedisTemplate rt : redisTemplates) {
    if (rt.opsForValue().setIfAbsent(lockKey, lockValue, lockExpireTime, TimeUnit.SECONDS)) {
        successCount++;
    }
}
// 验证是否获取多数节点锁
boolean locked = successCount >= redisTemplates.size() / 2 + 1;
long costTime = System.currentTimeMillis() - startTime;
// 有效时间需减去获取锁耗时
if (locked && costTime < lockExpireTime * 1000) {
    // 锁获取成功
} else {
    // 释放已获取的锁
    for (RedisTemplate rt : redisTemplates) {
        rt.execute(unlockScript, Collections.singletonList(lockKey), lockValue);
    }
}
// 需要至少3个独立的Redis实例
List<RedisTemplate> redisTemplates = Arrays.asList(redis1, redis2, redis3);

// 获取锁
String lockKey = "lock:order:123";
String lockValue = UUID.randomUUID().toString();
int lockExpireTime = 30; // 秒

int successCount = 0;
long startTime = System.currentTimeMillis();

for (RedisTemplate rt : redisTemplates) {
    if (rt.opsForValue().setIfAbsent(lockKey, lockValue, lockExpireTime, TimeUnit.SECONDS)) {
        successCount++;
    }
}

// 验证是否获取多数节点锁
boolean locked = successCount >= redisTemplates.size() / 2 + 1;
long costTime = System.currentTimeMillis() - startTime;

// 有效时间需减去获取锁耗时
if (locked && costTime < lockExpireTime * 1000) {
    // 锁获取成功
} else {
    // 释放已获取的锁
    for (RedisTemplate rt : redisTemplates) {
        rt.execute(unlockScript, Collections.singletonList(lockKey), lockValue);
    }
}
// 需要至少3个独立的Redis实例
List<RedisTemplate> redisTemplates = Arrays.asList(redis1, redis2, redis3);

// 获取锁
String lockKey = "lock:order:123";
String lockValue = UUID.randomUUID().toString();
int lockExpireTime = 30; // 秒

int successCount = 0;
long startTime = System.currentTimeMillis();

for (RedisTemplate rt : redisTemplates) {
    if (rt.opsForValue().setIfAbsent(lockKey, lockValue, lockExpireTime, TimeUnit.SECONDS)) {
        successCount++;
    }
}

// 验证是否获取多数节点锁
boolean locked = successCount >= redisTemplates.size() / 2 + 1;
long costTime = System.currentTimeMillis() - startTime;

// 有效时间需减去获取锁耗时
if (locked && costTime < lockExpireTime * 1000) {
    // 锁获取成功
} else {
    // 释放已获取的锁
    for (RedisTemplate rt : redisTemplates) {
        rt.execute(unlockScript, Collections.singletonList(lockKey), lockValue);
    }
}

特点：

• 需要部署多个独立Redis实例
• 遵循CP模型，牺牲部分性能换取可靠性
• 时钟漂移可能影响准确性

---

三、客户端框架方案

4. Redisson实现（生产级推荐）

// 配置Redisson客户端
Config config = new Config();
config.useSingleServer().setAddress("redis://127.0.0.1:6379");
RedissonClient redisson = Redisson.create(config);
// 获取锁对象
RLock lock = redisson.getLock("lock:order:123");
try {
    // 尝试加锁，最多等待10秒，锁定后30秒自动释放
    boolean isLock = lock.tryLock(10, 30, TimeUnit.SECONDS);
    if (isLock) {
        // 业务逻辑
    }
} finally {
    lock.unlock();
}
// 配置Redisson客户端
Config config = new Config();
config.useSingleServer().setAddress("redis://127.0.0.1:6379");
RedissonClient redisson = Redisson.create(config);

// 获取锁对象
RLock lock = redisson.getLock("lock:order:123");

try {
    // 尝试加锁，最多等待10秒，锁定后30秒自动释放
    boolean isLock = lock.tryLock(10, 30, TimeUnit.SECONDS);
    if (isLock) {
        // 业务逻辑
    }
} finally {
    lock.unlock();
}
// 配置Redisson客户端
Config config = new Config();
config.useSingleServer().setAddress("redis://127.0.0.1:6379");
RedissonClient redisson = Redisson.create(config);

// 获取锁对象
RLock lock = redisson.getLock("lock:order:123");

try {
    // 尝试加锁，最多等待10秒，锁定后30秒自动释放
    boolean isLock = lock.tryLock(10, 30, TimeUnit.SECONDS);
    if (isLock) {
        // 业务逻辑
    }
} finally {
    lock.unlock();
}

核心机制：

• 看门狗自动续期（默认30秒检测，续期到30秒）
• 可重入锁设计
• 支持公平锁和非公平锁
• 提供联锁（MultiLock）、红锁（RedLock）等扩展

---

四、特殊场景方案

5. 基于Hash的细粒度锁

// 对订单ID 123的库存项456加锁
String lockKey = "lock:order:123";
String field = "item_456";
// 加锁
redisTemplate.opsForHash().putIfAbsent(lockKey, field, "1");
redisTemplate.expire(lockKey, 30, TimeUnit.SECONDS);
// 解锁
redisTemplate.opsForHash().delete(lockKey, field);
// 对订单ID 123的库存项456加锁
String lockKey = "lock:order:123";
String field = "item_456";

// 加锁
redisTemplate.opsForHash().putIfAbsent(lockKey, field, "1");
redisTemplate.expire(lockKey, 30, TimeUnit.SECONDS);

// 解锁
redisTemplate.opsForHash().delete(lockKey, field);
// 对订单ID 123的库存项456加锁
String lockKey = "lock:order:123";
String field = "item_456";

// 加锁
redisTemplate.opsForHash().putIfAbsent(lockKey, field, "1");
redisTemplate.expire(lockKey, 30, TimeUnit.SECONDS);

// 解锁
redisTemplate.opsForHash().delete(lockKey, field);

适用场景：需要对同一资源的不同部分加锁

6. 乐观锁实现（CAS机制）

// 获取当前版本号
int currentVersion = Integer.parseInt(redisTemplate.opsForValue().get("version:order:123"));
// 业务处理...
// 更新时检查版本
redisTemplate.execute(new SessionCallback() {
    @Override
    public Object execute(RedisOperations operations) {
        operations.watch("version:order:123");
        int newVersion = Integer.parseInt(operations.opsForValue().get("version:order:123"));
        if (newVersion == currentVersion) {
            operations.multi();
            operations.opsForValue().increment("version:order:123");
            return operations.exec(); // 返回null表示失败
        }
        operations.unwatch();
        return null;
    }
});
// 获取当前版本号
int currentVersion = Integer.parseInt(redisTemplate.opsForValue().get("version:order:123"));

// 业务处理...

// 更新时检查版本
redisTemplate.execute(new SessionCallback() {
    @Override
    public Object execute(RedisOperations operations) {
        operations.watch("version:order:123");
        int newVersion = Integer.parseInt(operations.opsForValue().get("version:order:123"));
        if (newVersion == currentVersion) {
            operations.multi();
            operations.opsForValue().increment("version:order:123");
            return operations.exec(); // 返回null表示失败
        }
        operations.unwatch();
        return null;
    }
});
// 获取当前版本号
int currentVersion = Integer.parseInt(redisTemplate.opsForValue().get("version:order:123"));

// 业务处理...

// 更新时检查版本
redisTemplate.execute(new SessionCallback() {
    @Override
    public Object execute(RedisOperations operations) {
        operations.watch("version:order:123");
        int newVersion = Integer.parseInt(operations.opsForValue().get("version:order:123"));
        if (newVersion == currentVersion) {
            operations.multi();
            operations.opsForValue().increment("version:order:123");
            return operations.exec(); // 返回null表示失败
        }
        operations.unwatch();
        return null;
    }
});

---

五、方案对比分析

方案	可靠性	性能	实现复杂度	适用场景
SETNX+EXPIRE	★★☆	★★★	★☆☆	简单场景，低并发
SET扩展参数	★★★	★★★	★★☆	常规业务，中等并发
RedLock	★★★	★★☆	★★★	高可靠要求，金融场景
Redisson	★★★	★★☆	★☆☆	生产环境首选
Hash细粒度锁	★★☆	★★☆	★★☆	资源分段锁定
乐观锁	★☆☆	★★★	★★★	冲突少的读多写少场景

---

六、生产环境注意事项

1. 锁续期机制：

   // Redisson看门狗实现原理
   private void scheduleExpirationRenewal(long threadId) {
       Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
           @Override
           public void run(Timeout timeout) {
               // 异步续期
               renewExpirationAsync(threadId);
           }
       }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
   }
   // Redisson看门狗实现原理
   private void scheduleExpirationRenewal(long threadId) {
       Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
           @Override
           public void run(Timeout timeout) {
               // 异步续期
               renewExpirationAsync(threadId);
           }
       }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
   }
   // Redisson看门狗实现原理
   private void scheduleExpirationRenewal(long threadId) {
       Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
           @Override
           public void run(Timeout timeout) {
               // 异步续期
               renewExpirationAsync(threadId);
           }
       }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
   }

2. 锁等待策略优化：

• 指数退避重试
• 熔断机制防止雪崩

2. 锁监控：

   # Redis监控命令
   redis-cli --bigkeys
   redis-cli monitor | grep 'lock:'
   # Redis监控命令
   redis-cli --bigkeys
   redis-cli monitor | grep 'lock:'
   # Redis监控命令
   redis-cli --bigkeys
   redis-cli monitor | grep 'lock:'

4. 死锁预防：

• 设置合理的TTL
• 实现锁的自动释放
• 添加线程标识防止误删

---

七、选型建议

1. 中小规模系统：直接使用Redisson框架
2. 高并发场景：RedLock + 本地缓存降级
3. 秒杀系统：Redis锁 + 库存分段 + 乐观锁
4. 旧版本Redis：Lua脚本实现原子操作

正确实现分布式锁需要综合考虑CAP理论中的权衡，建议在开发完成后进行：

• 并发压力测试
• 网络分区模拟
• 故障恢复演练

最终方案应根据业务特点、Redis版本和团队技术栈综合决定。