当前位置： 首页 > web >正文

分布式锁的几种实现

web 2025/5/2 5:29:47

前几天看一个面试视频，提到了分布式锁一直想写写，但奈何考试太多，直到今天才有时间。好啦，开始今天的文章吧。

一.定义

分布式锁：当多个进程不在同一个系统中(比如分布式系统中控制共享资源访问)，用分布式锁控制多个进程对资源的访问。

二.实现

  • 基于数据库实现分布式锁  主要时依赖数据库支持的锁实现的

全局锁，表锁(排他锁，共享锁) ，行锁(记录锁，间隙锁，临键锁)，意向锁，悲观锁，乐观锁

基于数据库表  可以通过唯一索引实现，我给一个唯一索引的字段插入数据时，其他线程是无法给这个字段继续插入数据(唯一索引特性)，从而保证安全与一直性。

同样也可以通过主键索引实现(唯一+非空约束)   同上插入数据时相当于加锁，删除数据时相当于释放锁。

乐观锁(基于版本号)

如果存在一条数据插入时，另一条数据也要操作，就会检查版本号有没有更新，更新就再次获取锁，没有就插入数据。(如下为转账操作)

CREATE TABLE account (id INT PRIMARY KEY,balance DECIMAL(10,2),version INT NOT NULL
);
public boolean transferWithOptimisticLock(int id, double amount) throws SQLException {Connection conn = dataSource.getConnection();conn.setAutoCommit(false);try {// 查询当前余额与版本号String selectSql = "SELECT balance, version FROM account WHERE id = ?";PreparedStatement selectStmt = conn.prepareStatement(selectSql);selectStmt.setInt(1, id);ResultSet rs = selectStmt.executeQuery();if (!rs.next()) return false;double currentBalance = rs.getDouble("balance");int currentVersion = rs.getInt("version");if (currentBalance < amount) {throw new RuntimeException("余额不足");}double newBalance = currentBalance - amount;int newVersion = currentVersion + 1;// 更新并检查版本号String updateSql = "UPDATE account SET balance = ?, version = ? WHERE id = ? AND version = ?";PreparedStatement updateStmt = conn.prepareStatement(updateSql);updateStmt.setDouble(1, newBalance);updateStmt.setInt(2, newVersion);updateStmt.setInt(3, id);updateStmt.setInt(4, currentVersion);int rowsAffected = updateStmt.executeUpdate();if (rowsAffected == 0) {// 版本不一致，说明其他人已修改System.out.println("乐观锁失败：数据已被修改，请重试");return false;}conn.commit();return true;} catch (SQLException e) {conn.rollback();throw e;} finally {conn.setAutoCommit(true);}
}

悲观锁(基于排它锁)

一条数据插入时就直接加锁，然后等到数据插入完成就释放锁，持有锁期间不允许又任何读写操作作用于该数据。

CREATE TABLE account (id INT PRIMARY KEY,balance DECIMAL(10,2)
);
public void transferWithPessimisticLock(int id, double amount) throws SQLException {Connection conn = dataSource.getConnection();conn.setAutoCommit(false);try {// 加锁查询（FOR UPDATE）String lockSql = "SELECT balance FROM account WHERE id = ? FOR UPDATE";PreparedStatement lockStmt = conn.prepareStatement(lockSql);lockStmt.setInt(1, id);ResultSet rs = lockStmt.executeQuery();if (!rs.next()) throw new RuntimeException("账户不存在");double currentBalance = rs.getDouble("balance");if (currentBalance < amount) {throw new RuntimeException("余额不足");}double newBalance = currentBalance - amount;// 执行更新String updateSql = "UPDATE account SET balance = ? WHERE id = ?";PreparedStatement updateStmt = conn.prepareStatement(updateSql);updateStmt.setDouble(1, newBalance);updateStmt.setInt(2, id);updateStmt.executeUpdate();conn.commit();} catch (SQLException e) {conn.rollback();throw e;} finally {conn.setAutoCommit(true);}
}
  • 基于 redis 实现分布式锁:(常用)

单个Redis实例：setnx(key,当前时间+过期时间) + Lua脚本

这个太常见了，黑马外卖不就是通过Redis结合防重Token和Lua脚本来实现幂等性校验。lua脚本保证了读取用户token与删除token是一步完成的。

Maven导入：

<dependency><groupId>redis.clients</groupId><artifactId>jedis</artifactId><version>4.0.1</version>
</dependency>

代码实现:

import redis.clients.jedis.Jedis;
import redis.clients.jedis.params.SetParams;import java.util.Collections;
import java.util.UUID;public class RedisDistributedLock {private final Jedis jedis;private final String lockKey;private final String requestId;private final int expireTime; // 毫秒public RedisDistributedLock(Jedis jedis, String lockKey, int expireTime) {this.jedis = jedis;this.lockKey = lockKey;this.expireTime = expireTime;this.requestId = UUID.randomUUID().toString(); // 唯一标识}/*** 获取锁（带自动过期）*/public boolean acquire() {SetParams params = new SetParams();params.nx();          // 仅当 key 不存在时才设置params.px(expireTime); // 设置过期时间（毫秒）String result = jedis.set(lockKey, requestId, params);return "OK".equals(result);}/*** 释放锁（通过 Lua 脚本保证原子性）*/public boolean release() {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(requestId));return "1".equals(result.toString());}public static void main(String[] args) {try (Jedis jedis = new Jedis("localhost", 6379)) {RedisDistributedLock lock = new RedisDistributedLock(jedis, "my:lock:key", 30000);if (lock.acquire()) {System.out.println("【线程：" + Thread.currentThread().getName() + "】获取锁成功");try {// 模拟业务逻辑Thread.sleep(5000);} catch (InterruptedException e) {e.printStackTrace();} finally {if (lock.release()) {System.out.println("【线程：" + Thread.currentThread().getName() + "】释放锁成功");} else {System.out.println("【线程：" + Thread.currentThread().getName() + "】释放锁失败，可能已被他人释放");}}} else {System.out.println("【线程：" + Thread.currentThread().getName() + "】获取锁失败");}} catch (Exception e) {e.printStackTrace();}}
}

Redis集群模式：Redlock(红锁，存在争议)

原理：

  • 使用 N 个独立的 Redis 节点 （推荐为奇数个，如 5 个）
  • 每个节点都尝试获取相同的锁
  • 客户端通过多数节点加锁成功来判断是否获得锁

4. 关键问题与争议

问题

描述

时间依赖

Redlock 假设时间是同步的，但现实中 NTP 或虚拟机暂停会导致时间跳跃

锁安全性

Martin Kleppmann 指出，在某些异常场景下可能多个客户端同时持有锁

实用性

Antirez 强调 Redlock 更适合工程实践而非理论完美性

Maven导入：

<dependency><groupId>org.redisson</groupId><artifactId>redisson</artifactId><version>3.20.1</version>
</dependency>

代码实现(基于Redission):

import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;import java.util.concurrent.TimeUnit;public class RedlockDemo {public static void main(String[] args) {// 配置多个 Redis 节点（模拟集群）Config config1 = new Config();config1.useSingleServer().setAddress("redis://127.0.0.1:6379");Config config2 = new Config();config2.useSingleServer().setAddress("redis://127.0.0.1:6380");Config config3 = new Config();config3.useSingleServer().setAddress("redis://127.0.0.1:6381");// 创建三个 Redisson 客户端实例RedissonClient redisson1 = Redisson.create(config1);RedissonClient redisson2 = Redisson.create(config2);RedissonClient redisson3 = Redisson.create(config3);// 获取每个节点上的锁对象RLock lock1 = redisson1.getLock("redlock:key");RLock lock2 = redisson2.getLock("redlock:key");RLock lock3 = redisson3.getLock("redlock:key");// 创建 Redlock 对象RLock redLock = redisson1.getRedLock(lock1, lock2, lock3);boolean isLocked = false;try {// 尝试加锁，等待最多 100 秒，上锁后 30 秒自动解锁isLocked = redLock.tryLock(100, 30, TimeUnit.SECONDS);if (isLocked) {System.out.println("【加锁成功】当前线程：" + Thread.currentThread().getName());// 执行业务逻辑Thread.sleep(5000);} else {System.out.println("【加锁失败】");}} catch (InterruptedException e) {e.printStackTrace();} finally {if (isLocked) {redLock.unlock(); // 释放锁System.out.println("【锁已释放】");}// 关闭客户端redisson1.shutdown();redisson2.shutdown();redisson3.shutdown();}}
}
  • 基于 zookeeper实现分布式锁(zookeeper相对与redis而言舍弃了部分高性能而保证了强一致性)

临时有序节点来实现的分布式锁,Curator

核心思想：

  1. 每个客户端在 ZooKeeper 中尝试创建一个临时有序子节点 ，如：/locks/my_lock/lock_0000000001
  2. 获取当前所有子节点并排序，判断自己是否是最小序号的节点：
    • 如果是 → 成功获取锁；
    • 如果不是 → 监听前一个节点（Watch），等待它被删除；
  3. 执行完业务逻辑后，删除自己的节点 → 释放锁；
  4. 利用 ZooKeeper 的 Watcher 机制自动通知下一个节点尝试加锁；
String ourPath = zk.createEphemeralSequential(root + "/lock_", data);
List<String> children = zk.getChildren(root);
Collections.sort(children); // 排序所有子节点
//伪代码
if (ourPath is the first in sorted list) {return true; // 成功获取锁
} else {String prevNode = findPrevNode(ourPath, children);watch(prevNode); // 监听前一个节点waitUntilPrevNodeDeleted(); // 等待删除后再次尝试
}

Maven导入：

<dependency><groupId>org.apache.curator</groupId><artifactId>curator-framework</artifactId><version>5.7.0</version>
</dependency><dependency><groupId>org.apache.curator</groupId><artifactId>curator-recipes</artifactId><version>5.7.0</version>
</dependency>

代码实现:

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public class ZookeeperDistributedLockExample {// ZooKeeper 地址（单机模式）private static final String ZK_ADDRESS = "localhost:2181";// 锁路径（可以理解为资源名）private static final String LOCK_PATH = "/distributed_lock";public static void main(String[] args) {// 创建 ZooKeeper 客户端CuratorFramework client = CuratorFrameworkFactory.newClient(ZK_ADDRESS,new ExponentialBackoffRetry(1000, 3));client.start();// 创建分布式锁对象InterProcessMutex lock = new InterProcessMutex(client, LOCK_PATH);// 模拟多个线程并发请求锁ExecutorService executor = Executors.newFixedThreadPool(5);for (int i = 0; i < 5; i++) {final int threadNum = i;executor.submit(() -> {try {System.out.println("线程 " + threadNum + " 正在尝试获取锁...");if (lock.acquire(10, TimeUnit.SECONDS)) { // 等待最多10秒try {System.out.println("线程 " + threadNum + " 成功获得锁");Thread.sleep(3000); // 模拟业务处理} finally {lock.release(); // 释放锁System.out.println("线程 " + threadNum + " 已释放锁");}} else {System.out.println("线程 " + threadNum + " 获取锁失败");}} catch (Exception e) {e.printStackTrace();}});}executor.shutdown();}
}
  • 基于 Consul 实现分布式锁

核心思想：

  1. 每个客户端尝试在 Consul KV 中创建一个键，并带上当前会话（Session）；
  2. Consul 使用 acquire 原子操作保证只有第一个成功设置该 key 的客户端才能获得锁；
  3. 如果 key 已存在且关联了一个未过期的 session，则其他客户端无法获取锁；
  4. 当持有锁的客户端释放锁或 session 失效时，key 被清除，其他客户端可以重新竞争；

注：以下代码需要实现启动 Consul Agent

import java.io.*;
import java.net.HttpURLConnection;
import java.net.URL;
import java.nio.charset.StandardCharsets;public class ConsulDistributedLock {private static final String CONSUL_URL = "http://localhost:8500/v1/kv/";private final String lockKey;     // 锁路径，如 "/locks/my_resource"private final String sessionId;   // 当前会话 IDprivate final String serviceName; // 客户端标识public ConsulDistributedLock(String lockKey, String serviceName) throws Exception {this.lockKey = lockKey;this.serviceName = serviceName;this.sessionId = createSession();}/*** 创建一个 Session*/private String createSession() throws Exception {String sessionJson = String.format("{\"Name\":\"%s\",\"TTL\":\"15s\"}", serviceName);String response = sendPost("http://localhost:8500/v1/session/create", sessionJson);return response.split("\"")[3]; // 简单提取 session ID}/*** 尝试获取锁*/public boolean acquire() throws IOException {String urlStr = CONSUL_URL + lockKey + "?acquire=" + sessionId;String payload = "locked-by:" + serviceName;String result = sendPut(urlStr, payload);return Boolean.parseBoolean(result);}/*** 释放锁*/public boolean release() throws IOException {String urlStr = CONSUL_URL + lockKey + "?release=" + sessionId;String payload = "locked-by:" + serviceName;String result = sendPut(urlStr, payload);return Boolean.parseBoolean(result);}/*** 心跳保持 Session 活跃*/public void renewSession() throws IOException {String urlStr = "http://localhost:8500/v1/session/renew/" + sessionId;sendPut(urlStr, "");}/*** 发送 PUT 请求*/private String sendPut(String urlStr, String body) throws IOException {return sendRequest(urlStr, "PUT", body);}/*** 发送 POST 请求*/private String sendPost(String urlStr, String body) throws IOException {return sendRequest(urlStr, "POST", body);}/*** 发送 HTTP 请求通用方法*/private String sendRequest(String urlStr, String method, String body) throws IOException {URL url = new URL(urlStr);HttpURLConnection conn = (HttpURLConnection) url.openConnection();conn.setRequestMethod(method);conn.setDoOutput(true);try (OutputStream os = conn.getOutputStream()) {byte[] input = body.getBytes(StandardCharsets.UTF_8);os.write(input, 0, input.length);}StringBuilder response = new StringBuilder();try (BufferedReader br = new BufferedReader(new InputStreamReader(conn.getInputStream(), StandardCharsets.UTF_8))) {String responseLine;while ((responseLine = br.readLine()) != null) {response.append(responseLine.trim());}}return response.toString();}public static void main(String[] args) throws Exception {ConsulDistributedLock lock = new ConsulDistributedLock("/locks/my_lock", "service-A");System.out.println("尝试获取锁...");if (lock.acquire()) {try {System.out.println("【线程：" + Thread.currentThread().getName() + "】获取锁成功");// 模拟业务逻辑执行for (int i = 0; i < 10; i++) {lock.renewSession(); // 保持心跳Thread.sleep(5000);}} finally {lock.release();System.out.println("锁已释放");}} else {System.out.println("获取锁失败");}}
}

感谢你看到这里，喜欢的可以点点关注哦。

http://www.xdnf.cn/news/3365.html

相关文章：

  • 使用HunyuanVideo搭建文本生视频大模型
  • OpenSSL应用实践：嵌入式数据安全实战指南
  • 使用Node编写轻量级后端快速入门
  • 极简GIT使用
  • 【内存管理】对象树（内存管理）
  • (持续更新)Ubuntu搭建LNMP（Linux + Nginx + MySQL + PHP）环境
  • DeepSeek生成Word文档的创新路径与应用
  • 【计算机视觉】三维视觉：Nerfstudio：模块化神经辐射场框架的技术突破与实战指南
  • 操作系统OS是如何指挥外围设备的呢？
  • MySQL 服务搭建
  • Leetcode刷题记录25——合并区间
  • MySQL与分布式架构的碰撞
  • 使用MGeo模型高精度实现文本中地址识别
  • 题解：洛谷 CF2091E Interesting Ratio
  • Java 中使用正则表达式
  • 在Linux中，KVM和Docker在Linux虚拟化中的区别是什么？
  • 【计算机视觉】语义分割：Mask2Former：统一分割框架的技术突破与实战指南
  • Mysql常用函数解析
  • Annotate better with CVAT
  • 华为OD机试真题——斗地主之顺子（2025A卷：100分）Java/python/JavaScript/C/C++/GO最佳实现
  • 《TCP/IP详解 卷1：协议》之第九章：IP选路
  • 湖仓一体化介绍
  • 数据库基本概念：数据库的定义、特点、分类、组成、作用
  • 详解TypeScript中的类型断言及其绕过类型检查机制
  • 【Vue bug】:deep()失效
  • 如何提升自我执行力？
  • 拆解 browser-use 项目——深入理解 Agent 层
  • Linux 环境下 Mysql 5.7 数据定期备份
  • Kotlin-运算符重载函数
  • 生产级RAG系统一些经验总结