前言

在分布式系统中，并发控制就像是多个人同时进出一个房间，如何确保秩序？

想象一下这个场景：双十一抢购活动中，某个热门商品只剩最后 10 件，此时有几万人同时点击"下单"。如果没有适当的并发控制，很可能导致最终售出 100 件（严重超卖），但实际仓库只有 10 件商品，这将带来糟糕的用户体验和业务混乱。

在单机应用中，我们可以使用 Java 的 synchronized 关键字或 ReentrantLock 轻松解决这类问题。但在分布式环境下，这些本地锁无法跨服务器工作，此时我们需要一种能在多个服务实例间协调的锁机制——分布式锁。

这让我想到了现实生活中的钥匙与门锁：只有持有钥匙的人才能开门，而一把钥匙同一时间只能由一个人持有。分布式锁正是这一概念在计算机世界的实现。

为什么选择 Redis 实现分布式锁？

实现分布式锁的技术方案有很多，包括数据库、Zookeeper、Redis 等。在经过反复比较后，我认为 Redis 是实现分布式锁的绝佳选择，原因如下：

1. 性能卓越：Redis 的内存操作使其响应时间通常在亚毫秒级，每秒可处理数万次锁操作
2. 实现简单：只需几个命令即可实现基本的分布式锁功能
3. 部署友好：相比 Zookeeper，Redis 的部署和维护成本更低
4. 自动过期：支持锁的自动过期机制，有效防止死锁
5. 普及度高：大多数系统已经使用 Redis 做缓存，无需引入新组件

当然，万事没有绝对。在某些要求极高一致性的场景下，Zookeeper 可能是更好的选择。但对于大多数应用场景，Redis 提供了最佳的性能与可靠性平衡。

Redis 分布式锁的核心原理

我认为 Redis 分布式锁的精妙之处在于利用了 Redis 的原子操作特性。它的核心原理可以用"占位子"来类比：

想象一个会议室只有一把椅子，谁先到会议室并坐在椅子上，谁就拥有了发言权（获得锁）。直到这个人起身离开（释放锁），其他人才能坐下。

具体实现上，Redis 分布式锁涉及三个基本步骤：

1. 获取锁：尝试在 Redis 中创建一个不存在的键，成功则获得锁
2. 持有锁：获得锁的客户端执行其业务逻辑
3. 释放锁：操作完成后删除该键，或等待其自动过期

实现一个可靠的分布式锁

在尝试了多种方案后，我总结出实现可靠分布式锁的关键代码。

获取锁

首先，加锁操作需要使用 Redis 的 SET 命令配合特定选项：

# NX: 只在键不存在时设置（互斥性）
# PX: 设置过期时间（毫秒，防止死锁）
SET lock:product:10001 client-12345 NX PX 10000

• 通过 NX 选项确保互斥性
• 通过设置值为客户端唯一标识，后续可以验证锁的所有权
• 通过 PX 设置过期时间，防止客户端崩溃时锁无法释放

释放锁

释放锁看似简单，但需要注意一个细节：确保只释放自己的锁。为此，我们需要使用 Lua 脚本来保证检查和删除的原子性：

if redis.call('get', KEYS[1]) == ARGV[1] then
    return redis.call('del', KEYS[1])
else
    return 0
end

这个脚本检查锁的拥有者是否为当前客户端，只有匹配时才删除锁。

没看不懂或者没用 Lua 脚本没关系，只要知道这是一个判断删除操作即可。

Java 实现分布式锁

以下为学习用的 Demo 工程代码，仅供参考。

一个基于 Spring Boot 的实现，我尽量使其简洁：

@Service
public class RedisLockService {

    @Autowired
    private StringRedisTemplate redisTemplate;

    private static final long DEFAULT_LOCK_TIMEOUT = 10000; // 10秒
    private static final String UNLOCK_SCRIPT =
        "if redis.call('get', KEYS[1]) == ARGV[1] then " +
        "return redis.call('del', KEYS[1]) else return 0 end";

    /**
     * 尝试获取锁
     */
    public boolean tryLock(String lockKey, String clientId, long timeout) {
        Boolean success = redisTemplate.opsForValue().setIfAbsent(
            lockKey, clientId, timeout, TimeUnit.MILLISECONDS);
        return Boolean.TRUE.equals(success);
    }

    /**
     * 释放锁（安全的，只释放自己的锁）
     */
    public boolean releaseLock(String lockKey, String clientId) {
    // 执行 Lua 脚本，原子性检查锁的持有者是否为 clientId，若是则删除
    Long result = redisTemplate.execute(
        new DefaultRedisScript<>(UNLOCK_SCRIPT, Long.class), // 封装 Lua 脚本和返回类型
        Collections.singletonList(lockKey), // 脚本参数：key 列表
        clientId // 脚本参数：ARGV[1]，当前客户端 ID
    );
    // 判断脚本执行结果是否为 1（1 表示成功删除锁）
    return Long.valueOf(1).equals(result);
}

    /**
     * 执行带锁的操作
     */
    public <T> T executeWithLock(String lockKey, Supplier<T> task) {
        String clientId = UUID.randomUUID().toString();
        boolean locked = tryLock(lockKey, clientId, DEFAULT_LOCK_TIMEOUT);

        if (!locked) {
            throw new RuntimeException("获取锁失败: " + lockKey);
        }

        try {
            return task.get();
        } finally {
            releaseLock(lockKey, clientId);
        }
    }

    /**
     * 执行带锁的操作（无返回值版本）
     */
    public void executeWithLock(String lockKey, Runnable task) {
        executeWithLock(lockKey, () -> {
            task.run();
            return null;
        });
    }
}

分布式锁的典型应用场景

分布式锁在多种场景下非常有用，下面用一些经典例子：

库存管理与防止超卖

@Service
public class StockService {

    @Autowired
    private RedisLockService lockService;

    @Autowired
    private ProductRepository productRepository;

    /**
     * 扣减库存（防止超卖）
     */
    public boolean decreaseStock(Long productId, int quantity) {
        String lockKey = "lock:stock:" + productId;

        return lockService.executeWithLock(lockKey, () -> {
            // 查询当前库存
            Product product = productRepository.findById(productId)
                .orElseThrow(() -> new RuntimeException("商品不存在"));

            // 检查库存是否充足
            if (product.getStock() < quantity) {
                return false; // 库存不足
            }

            // 扣减库存
            product.setStock(product.getStock() - quantity);
            productRepository.save(product);
            return true;
        });
    }
}

防重复提交

@RestController
@RequestMapping("/api/orders")
public class OrderController {

    @Autowired
    private RedisLockService lockService;

    @Autowired
    private OrderService orderService;

    @PostMapping("/create")
    public ResponseEntity<?> createOrder(@RequestBody OrderRequest request) {
        // 生成幂等键，例如基于用户ID和请求参数的哈希
        String idempotentKey = "idempotent:" + request.getUserId() + ":"
                              + DigestUtils.md5DigestAsHex(request.toString().getBytes());

        try {
            return lockService.executeWithLock("lock:" + idempotentKey, () -> {
                // 检查是否已经处理过
                if (orderService.isProcessed(idempotentKey)) {
                    return ResponseEntity.ok("订单已处理，请勿重复提交");
                }

                // 创建订单
                Order order = orderService.createOrder(request);

                // 标记为已处理
                orderService.markAsProcessed(idempotentKey);

                return ResponseEntity.ok(order);
            });
        } catch (Exception e) {
            return ResponseEntity.status(HttpStatus.TOO_MANY_REQUESTS)
                    .body("请求正在处理中，请勿重复提交");
        }
    }
}

分布式定时任务控制

@Component
public class ScheduledTasks {

    @Autowired
    private RedisLockService lockService;

    /**
     * 每小时执行一次的任务
     */
    @Scheduled(cron = "0 0 * * * ?")
    public void hourlyTask() {
        String lockKey = "lock:scheduled:hourly-task";

        try {
            // 尝试获取锁，如果获取不到，说明有其他实例正在执行该任务
            lockService.executeWithLock(lockKey, () -> {
                System.out.println("执行小时级定时任务，时间：" + new Date());
                // 执行实际的定时任务...
            });
        } catch (Exception e) {
            // 获取锁失败，忽略本次执行
            System.out.println("其他服务实例正在执行该任务，本实例跳过");
        }
    }
}

最佳实践

在实际应用分布式锁的过程中，我总结了几点最佳实践：

合理设置锁超时时间

锁的超时时间应该略长于业务执行时间，避免业务未完成锁就过期的情况。举例来说，如果业务通常在 200 毫秒内完成，可以将锁超时设置为 1000 毫秒，留出足够的余量。

考虑锁续期

对于执行时间不确定的业务，可以实现"看门狗"机制，定期续期：

ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
scheduler.scheduleAtFixedRate(() -> {
    // 每5秒续期一次，将过期时间延长10秒
    redisTemplate.expire(lockKey, 10, TimeUnit.SECONDS);
}, 5, 5, TimeUnit.SECONDS);

try {
    // 执行业务逻辑
} finally {
    scheduler.shutdown();
    // 释放锁
}

规范锁的粒度

粒度过粗的锁会导致不必要的竞争，而粒度过细则会增加系统复杂度。例如，商品库存操作通常以单个商品为粒度加锁，而不是锁定整个库存表。

总结

Redis 基于 String 类型实现的分布式锁，为我们提供了一种简单而高效的并发控制方案。它适用于大多数分布式系统，尤其适合对性能要求较高的场景。

在实际应用中，除了掌握基本的实现方法外，更需要根据业务特性选择适当的锁粒度、超时策略和异常处理机制。同时，对于金融等要求极高一致性的场景，可能需要考虑使用 Redisson 的 RedLock 算法或 Zookeeper 等替代方案。

正如一位前辈所言：“选择合适的工具比精通所有工具更重要。” 希望这篇文章能帮助你在分布式系统中正确地应用锁机制，构建更可靠的应用程序。

值得深入思考的问题

通过实际项目中的应用，我发现分布式锁还存在一些值得深入思考的问题：

1. 锁的可重入性：如何实现分布式环境下的可重入锁？同一个客户端可能需要多次获取同一把锁，普通实现会导致自己锁死自己。
2. 锁的公平性：如何确保锁的获取是公平的？在高并发场景下，可能会出现某些客户端一直获取不到锁的"饥饿"现象。
3. 故障转移与脑裂问题：在 Redis 主从架构中，如果主节点宕机，从节点还没来得及同步锁信息就被提升为主节点，可能导致同一把锁被两个客户端同时持有。
4. 性能与可靠性的平衡：更可靠的分布式锁通常意味着更多的网络通信和延迟，如何在保证锁可靠性的同时不过分损失性能？
5. 降级策略：当 Redis 服务不可用时，系统如何优雅降级？是直接拒绝请求，还是转为本地锁或数据库锁？

后记

说实话，写这一篇文章难度好大，主要是我的日常很少接触到使用 Redis 作为分布式锁的场景。后面的代码基本都是跟着 AI 写的，哈哈。

附录

• Redis String 实现分布式锁