Java 分布式锁：原理与实践

news2024/9/22 10:42:08

在分布式系统中，多个节点同时操作共享资源的情况非常普遍。为了保证数据的一致性，分布式锁 应运而生。分布式锁 是一种跨多个服务器的互斥锁，用于协调分布式环境下的资源访问。

本文将介绍 Java 实现分布式锁 的几种常见方式，并结合 Redis、Zookeeper 等框架给出代码实例，探讨如何在实际项目中运用这些技术。

在这里插入图片描述

一、什么是分布式锁？

分布式锁的本质是保证在分布式环境下，多个进程或者节点可以按照正确的顺序来访问共享资源，避免数据冲突。具体场景包括：

限流控制：确保多个服务器上的相同任务不会被重复执行。
资源竞争：多个节点对数据库或文件等共享资源的并发修改。
任务调度：某些任务在同一时间只能由一个节点执行。

二、分布式锁的常见实现方式

常见的分布式锁实现方式有：

基于数据库的分布式锁
基于 Redis 的分布式锁
基于 Zookeeper 的分布式锁

下面我们分别介绍这些方法及其在 Java 中的实现。

三、基于数据库的分布式锁

1. 使用数据库表实现分布式锁

通过创建一张锁表，在锁定资源时插入一条记录，通过 唯一约束 来确保同一时间只有一个节点能成功插入数据。

代码示例：

CREATE TABLE distributed_lock (
    lock_name VARCHAR(64) PRIMARY KEY,
    locked_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

Java 实现：

import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;

@Service
public class DatabaseLockService {

    private final JdbcTemplate jdbcTemplate;

    public DatabaseLockService(JdbcTemplate jdbcTemplate) {
        this.jdbcTemplate = jdbcTemplate;
    }

    public boolean acquireLock(String lockName) {
        try {
            jdbcTemplate.update("INSERT INTO distributed_lock (lock_name) VALUES (?)", lockName);
            return true;
        } catch (Exception e) {
            return false; // 如果插入失败，表示锁已经被占用
        }
    }

    public void releaseLock(String lockName) {
        jdbcTemplate.update("DELETE FROM distributed_lock WHERE lock_name = ?", lockName);
    }
}

注意事项：

加锁失败的处理：如果锁已被占用，可能需要设置重试机制或返回失败。
锁超时机制：为了避免死锁，可以在数据库中设置锁的超时时间。

四、基于 Redis 的分布式锁

Redis 是分布式锁最常用的实现之一，主要依赖于 Redis 提供的 SETNX（SET if Not Exists）命令，来确保只有一个客户端能成功设置某个键值。

1. 使用 Redis 实现分布式锁

代码示例：

import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import java.util.concurrent.TimeUnit;

@Service
public class RedisLockService {

    private final StringRedisTemplate redisTemplate;

    public RedisLockService(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    public boolean acquireLock(String lockKey, String lockValue, long timeout) {
        Boolean success = redisTemplate.opsForValue().setIfAbsent(lockKey, lockValue, timeout, TimeUnit.SECONDS);
        return success != null && success;
    }

    public void releaseLock(String lockKey, String lockValue) {
        String currentValue = redisTemplate.opsForValue().get(lockKey);
        if (lockValue.equals(currentValue)) {
            redisTemplate.delete(lockKey); // 删除锁
        }
    }
}

注意事项：

过期时间：设置锁的过期时间，避免因某个节点宕机导致死锁。
锁的唯一标识：每个锁需要有一个唯一标识，释放锁时必须检查该标识是否一致，防止误删他人的锁。

2. 基于 Redisson 实现分布式锁

Redisson 是 Redis 官方推荐的 Java 客户端，提供了原生的分布式锁实现。

代码示例：

import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Service;

import java.util.concurrent.TimeUnit;

@Service
public class RedissonLockService {

    private final RedissonClient redissonClient;

    public RedissonLockService(RedissonClient redissonClient) {
        this.redissonClient = redissonClient;
    }

    public void lock(String lockKey) {
        RLock lock = redissonClient.getLock(lockKey);
        lock.lock(10, TimeUnit.SECONDS); // 获取锁并设置过期时间
    }

    public void unlock(String lockKey) {
        RLock lock = redissonClient.getLock(lockKey);
        if (lock.isHeldByCurrentThread()) {
            lock.unlock();
        }
    }
}

优点：

原生支持锁的自动续期：避免因超时导致的锁失效。
高性能：Redisson 提供了分布式锁的高效实现，适合高并发场景。

五、基于 Zookeeper 的分布式锁

Zookeeper 是一个分布式协调服务，天生支持分布式锁的实现。Zookeeper 中的 临时节点 和 顺序节点 是实现分布式锁的关键。

1. 使用 Zookeeper 实现分布式锁

通过 Zookeeper 的 临时顺序节点 实现锁机制，保证只有第一个创建的节点可以获取锁，其他节点需要监听前一个节点的删除事件。

代码示例：

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.Collections;
import java.util.List;

public class ZookeeperLockService {

    private final ZooKeeper zooKeeper;
    private String lockPath;

    public ZookeeperLockService(String zkAddress) throws IOException {
        zooKeeper = new ZooKeeper(zkAddress, 3000, null);
    }

    public boolean acquireLock(String lockName) throws Exception {
        lockPath = zooKeeper.create("/locks/" + lockName + "/lock_", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        List<String> children = zooKeeper.getChildren("/locks/" + lockName, false);
        Collections.sort(children);
        String smallestChild = children.get(0);
        if (lockPath.endsWith(smallestChild)) {
            return true; // 获取锁成功
        } else {
            // 等待前一个节点删除
            int previousNodeIndex = Collections.binarySearch(children, lockPath.substring(lockPath.lastIndexOf('/') + 1)) - 1;
            String previousNodePath = "/locks/" + lockName + "/" + children.get(previousNodeIndex);
            Stat stat = zooKeeper.exists(previousNodePath, new LockWatcher());
            return stat == null;
        }
    }

    public void releaseLock() throws Exception {
        zooKeeper.delete(lockPath, -1);
    }

    private class LockWatcher implements Watcher {
        @Override
        public void process(WatchedEvent event) {
            synchronized (this) {
                this.notifyAll(); // 通知等待的线程
            }
        }
    }
}