概述
在分布式系统中，多个进程或服务可能需要协调对共享资源的访问。分布式锁是解决这一问题的经典方案。本文将用Java实现基于Redis的分布式锁，包括单节点锁、Redlock红锁算法以及锁的续约与自动释放机制，帮助读者理解分布式环境下的协调原语。
核心挑战
分布式锁需要满足四个基本条件：
- 互斥性：任意时刻只有一个客户端持有锁
- 防死锁：即使持有锁的客户端崩溃，锁也能最终释放
- 容错性：只要大多数Redis节点正常，锁就能正常工作
- 一致性：锁的获取和释放需要原子操作保证
Java实现
单节点Redis锁
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.params.SetParams;
import java.util.Collections;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
public class RedisDistributedLock {
private final JedisPool jedisPool;
private final String lockKey;
private final String lockValue;
private final int expireSeconds;
public RedisDistributedLock(JedisPool pool, String lockKey, int expireSeconds) {
this.jedisPool = pool;
this.lockKey = lockKey;
this.expireSeconds = expireSeconds;
// 使用UUID保证每个客户端的锁标识唯一
this.lockValue = UUID.randomUUID().toString() + "-" + Thread.currentThread().getId();
}
/**
* 尝试获取锁（非阻塞）
* 使用 SET key value NX EX seconds 原子命令
*/
public boolean tryLock() {
try (Jedis jedis = jedisPool.getResource()) {
SetParams params = SetParams.setParams()
.nx() // 仅当key不存在时才设置
.ex(expireSeconds); // 设置过期时间
String result = jedis.set(lockKey, lockValue, params);
return "OK".equals(result);
}
}
/**
* 阻塞式获取锁，带超时
*/
public boolean lock(long waitTime, TimeUnit unit) throws InterruptedException {
long deadline = System.nanoTime() + unit.toNanos(waitTime);
while (System.nanoTime() < deadline) {
if (tryLock()) {
return true;
}
Thread.sleep(100); // 自旋间隔
}
return false;
}
/**
* 释放锁：使用Lua脚本保证原子性
* 仅当锁值匹配时才删除，防止误删其他客户端的锁
*/
public boolean unlock() {
String luaScript =
"if redis.call('get', KEYS[1]) == ARGV[1] then " +
" return redis.call('del', KEYS[1]) " +
"else " +
" return 0 " +
"end";
try (Jedis jedis = jedisPool.getResource()) {
Object result = jedis.eval(luaScript,
Collections.singletonList(lockKey),
Collections.singletonList(lockValue));
return Long.valueOf(1).equals(result);
}
}
public String getLockValue() {
return lockValue;
}
}
看门狗自动续约
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
public class WatchdogRenewal {
private final ScheduledExecutorService scheduler;
private final RedisDistributedLock lock;
private final int lockExpireSeconds;
private final AtomicBoolean running = new AtomicBoolean(false);
public WatchdogRenewal(RedisDistributedLock lock, int lockExpireSeconds) {
this.lock = lock;
this.lockExpireSeconds = lockExpireSeconds;
this.scheduler = new ScheduledThreadPoolExecutor(1);
}
/**
* 启动看门狗：每1/3过期时间续约一次
*/
public void start() {
if (running.compareAndSet(false, true)) {
long renewInterval = lockExpireSeconds * 1000 / 3;
scheduler.scheduleAtFixedRate(this::renew,
renewInterval, renewInterval, TimeUnit.MILLISECONDS);
}
}
/**
* 续约逻辑：检查锁是否仍属于自己，然后延长过期时间
*/
private void renew() {
String luaScript =
"if redis.call('get', KEYS[1]) == ARGV[1] then " +
" return redis.call('expire', KEYS[1], ARGV[2]) " +
"else " +
" return 0 " +
"end";
try (Jedis jedis = lock.getJedisPool().getResource()) {
Object result = jedis.eval(luaScript,
Collections.singletonList(lock.getLockKey()),
Arrays.asList(lock.getLockValue(), String.valueOf(lockExpireSeconds)));
if (!Long.valueOf(1).equals(result)) {
// 续约失败，锁可能已丢失
stop();
}
}
}
public void stop() {
if (running.compareAndSet(true, false)) {
scheduler.shutdown();
}
}
}
Redlock红锁算法
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;
public class RedLock {
private final List<JedisPool> redisNodes;
private final int quorum; // 多数派阈值 = N/2 + 1
private final int lockExpireMs;
public RedLock(List<JedisPool> nodes, int lockExpireMs) {
this.redisNodes = nodes;
this.quorum = nodes.size() / 2 + 1;
this.lockExpireMs = lockExpireMs;
}
/**
* Redlock算法核心步骤：
* 1. 获取当前时间戳
* 2. 依次向所有N个节点申请锁，使用相同的key和随机value
* 3. 计算获取锁的总耗时
* 4. 若成功获取锁的节点数 >= 多数派，且总耗时 < 锁过期时间，则加锁成功
* 5. 若失败，向所有节点发送释放锁请求
*/
public LockResult tryLock(String resourceName) {
String lockValue = generateUniqueValue();
long startTime = System.currentTimeMillis();
int successCount = 0;
List<JedisPool> lockedNodes = new ArrayList<>();
for (JedisPool node : redisNodes) {
try (Jedis jedis = node.getResource()) {
// 使用PX指定毫秒级过期时间
String result = jedis.set(resourceName, lockValue,
SetParams.setParams().nx().px(lockExpireMs));
if ("OK".equals(result)) {
successCount++;
lockedNodes.add(node);
}
} catch (Exception ex) {
// 单个节点故障不影响整体判断
}
// 随机延迟，避免所有客户端同时请求导致活锁
try {
Thread.sleep(ThreadLocalRandom.current().nextInt(5));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
long elapsed = System.currentTimeMillis() - startTime;
boolean locked = successCount >= quorum && elapsed < lockExpireMs;
if (!locked) {
// 加锁失败，释放已获取的锁
unlockOnAllNodes(resourceName, lockValue);
return new LockResult(false, null, null);
}
// 计算有效过期时间 = 原始过期时间 - 获取锁耗时 - 时钟漂移补偿（2ms）
long validityTime = lockExpireMs - elapsed - 2;
return new LockResult(true, lockValue, lockedNodes, validityTime);
}
/**
* 在所有节点上释放锁
*/
public void unlock(String resourceName, String lockValue, List<JedisPool> nodes) {
String luaScript =
"if redis.call('get', KEYS[1]) == ARGV[1] then " +
" return redis.call('del', KEYS[1]) " +
"else " +
" return 0 " +
"end";
for (JedisPool node : nodes) {
try (Jedis jedis = node.getResource()) {
jedis.eval(luaScript,
Collections.singletonList(resourceName),
Collections.singletonList(lockValue));
} catch (Exception ex) {
// 忽略释放失败的节点
}
}
}
private void unlockOnAllNodes(String resourceName, String lockValue) {
unlock(resourceName, lockValue, redisNodes);
}
private String generateUniqueValue() {
return UUID.randomUUID().toString() + "-" + System.nanoTime();
}
public static class LockResult {
public final boolean success;
public final String lockValue;
public final List<JedisPool> lockedNodes;
public final long validityTime;
public LockResult(boolean success, String lockValue,
List<JedisPool> lockedNodes, long validityTime) {
this.success = success;
this.lockValue = lockValue;
this.lockedNodes = lockedNodes;
this.validityTime = validityTime;
}
public LockResult(boolean success, String lockValue, List<JedisPool> lockedNodes) {
this(success, lockValue, lockedNodes, 0);
}
}
}
基于Redlock的分布式计数器
public class DistributedCounter {
private final RedLock redLock;
private final String counterKey;
private final JedisPool masterPool;
public DistributedCounter(RedLock redLock, String counterKey, JedisPool masterPool) {
this.redLock = redLock;
this.counterKey = counterKey;
this.masterPool = masterPool;
}
/**
* 原子递增：先获取Redlock，再执行Redis INCR
*/
public long increment() {
RedLock.LockResult lock = redLock.tryLock(counterKey + ":lock");
if (!lock.success) {
throw new RuntimeException("无法获取分布式锁");
}
try {
// 在有效时间内完成操作
long deadline = System.currentTimeMillis() + lock.validityTime;
try (Jedis jedis = masterPool.getResource()) {
long newValue = jedis.incr(counterKey);
if (System.currentTimeMillis() > deadline) {
throw new RuntimeException("锁已过期，操作可能不安全");
}
return newValue;
}
} finally {
redLock.unlock(counterKey + ":lock", lock.lockValue, lock.lockedNodes);
}
}
}
算法分析
容错能力
| 总节点数N | 多数派阈值 | 可容忍故障 |
|---|---|---|
| 3 | 2 | 1 |
| 5 | 3 | 2 |
| 7 | 4 | 3 |
时间复杂度
| 操作 | 时间复杂度 |
|---|---|
| 单节点加锁 | O(1) |
| Redlock加锁 | O(N) N为Redis节点数 |
| 解锁 | O(N) |
| 看门狗续约 | O(1) 每轮 |
实际应用建议
- 时钟同步：Redlock假设各节点时钟大致同步，最大漂移应远小于锁过期时间
- 锁续期：业务执行时间不确定时，务必使用看门狗机制
- 锁粒度：锁的粒度应尽可能小，减少竞争
- 重试策略：获取锁失败时采用指数退避，避免活锁
- 熔断机制：连续获取锁失败时应触发熔断，保护系统
总结
本文实现了从单节点Redis锁到Redlock红锁的完整分布式锁方案，涵盖了原子加锁、Lua脚本安全释放、看门狗自动续约等核心机制。分布式锁的本质是在无共享内存的环境中，通过外部协调服务实现互斥原语。理解其原理对于构建可靠的分布式系统至关重要。
