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

分布式锁的三种实现方式：Redis、基于数据库和Zookeeper

news 来源：原创 2024/9/20 9:43:29

分布式锁的实现

操作共享资源：例如操作数据库中的唯一用户数据、订单系统、优惠券系统、积分系统等，这些系统需要修改用户数据，而多个系统可能同时修改同一份数据，这时就需要使用分布式锁来控制访问，防止数据不一致。
在电商系统中，如果多个用户同时购买同一商品，可能会出现超卖现象。通过使用分布式锁，可以确保在同一时间只有一个用户能够进行购买操作，从而避免库存超卖的问题。‌
防止重复调用第三方接口：在分布式系统中，如果多个节点同时调用同一个第三方接口，可能会导致接口调用失败或数据错误。使用分布式锁可以确保在同一时间只有一个节点进行接口调用，避免重复调用问题。

转自：https://lingkang.top/archives/lock333

Redis

当客户端需要获取锁时，向Redis发送SETNX命令，如果返回1，说明客户端获得了锁；如果返回0，则说明锁已被其他客户端占用。当客户端释放锁时，使用DEL命令删除对应的键即可。

1、Maven中添加依赖

<dependency><groupId>cn.hutool</groupId><artifactId>hutool-core</artifactId><version>5.8.27</version>
</dependency>
<dependency><groupId>redis.clients</groupId><artifactId>jedis</artifactId><version>2.9.3</version>
</dependency>

2、启动好Redis

在这里插入图片描述

window可以在这里下载一个：https://gitee.com/lingkang_top/redis-window

3、编写java代码

package redis;import cn.hutool.core.thread.ThreadUtil;
import redis.clients.jedis.Jedis;import java.util.ArrayList;
import java.util.List;/*** @author lingkang* @create by 2024/7/25 15:28*/
public class Demo01 {/*** 假设是库存*/private static int number = 100;public static void main(String[] args) {// 提前初始化好redis连接List<Jedis> redis = new ArrayList<>();for (int i = 0; i < 10; i++) {Jedis jedis = new Jedis("localhost", 6379);redis.add(jedis);}// 操作前先清空锁redis.get(0).del("lock");// 假设有10个线程进行锁操作for (int i = 0; i < 10; i++) {final Jedis jedis = redis.get(i);new Thread(() -> {// 每隔线程减扣 5次for (int j = 0; j < 5; j++)dd(jedis);jedis.close();}, "t-" + i).start();}ThreadUtil.sleep(15000);// 应该输出 100-50=50System.out.println("库存: " + number);}/*** 业务处理*/private static void dd(Jedis jedis) {for (; ; ) {long lock = jedis.setnx("lock", "1");if (lock == 1) {// 给key设置一个过期时间，防止死锁jedis.expire("lock", 20);// 获取到锁System.out.println("当前线程获得锁：" + Thread.currentThread().getName());// 进行减扣库存等一系列操作....number--;// 假设处理业务延迟一下ThreadUtil.sleep(200);// 处理完毕要移除锁jedis.del("lock");break;}// 等待一下ThreadUtil.sleep(200);}}
}

4、结果正确

在这里插入图片描述

基于数据库

基于数据库的分布式锁主要依赖于数据库的唯一索引或主键约束。具体实现时，当客户端需要获取锁时，向数据库中插入一条记录，该记录的唯一键表示锁。如果插入成功，说明客户端获得了锁；如果插入失败（如主键冲突），则说明锁已被其他客户端占用。当客户端释放锁时，删除该记录即可。

1、Maven中添加依赖

<dependency><groupId>cn.hutool</groupId><artifactId>hutool-core</artifactId><version>5.8.27</version>
</dependency>
<dependency><groupId>com.mysql</groupId><artifactId>mysql-connector-j</artifactId><version>9.0.0</version>
</dependency>

2、添加表格

CREATE TABLE `mylock` (`id` char(1) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,`create_time` timestamp NULL DEFAULT NULL,PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;

3、编写java代码

package mysql;import cn.hutool.core.io.IoUtil;
import cn.hutool.core.thread.ThreadUtil;import java.sql.*;
import java.util.ArrayList;
import java.util.List;/*** @author lingkang* @create by 2024/7/25 16:58*/
public class Demo01 {/*** 假设是库存*/private static int number = 100;public static void main(String[] args) throws Exception {Class.forName("com.mysql.cj.jdbc.Driver");String URL = "jdbc:mysql://localhost:3306/mylock?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=Asia/Shanghai";// 提前初始化好redis连接List<Connection> connectionList = new ArrayList<>();for (int i = 0; i < 10; i++) {Connection conn = DriverManager.getConnection(URL, "root", "123456");connectionList.add(conn);}// 开始前，先将表表的锁数据清理connectionList.get(0).prepareStatement("delete from mylock where id='1'").executeUpdate();// 假设有10个线程进行锁操作for (int i = 0; i < 10; i++) {final Connection conn = connectionList.get(i);new Thread(() -> {// 每隔线程减扣 5次for (int j = 0; j < 5; j++) {try {dd(conn);} catch (Exception e) {throw new RuntimeException(e);}}IoUtil.close(conn);}, "t-" + i).start();}ThreadUtil.sleep(15000);// 应该输出 100-50=50System.out.println("库存: " + number);}/*** 业务处理*/private static void dd(Connection conn) throws Exception {for (; ; ) {int success = 0;try {PreparedStatement statement = conn.prepareStatement("insert into mylock(id,create_time) values('1',now())");success = statement.executeUpdate();} catch (Exception e) {}if (success == 1) {// 获取到锁System.out.println("当前线程获得锁：" + Thread.currentThread().getName());// 进行减扣库存等一系列操作....number--;// 假设处理业务延迟一下ThreadUtil.sleep(200);// 处理完毕要移除锁PreparedStatement statement = conn.prepareStatement("delete from mylock where id='1'");statement.executeUpdate();statement.close();break;} else {ResultSet query = conn.prepareStatement("select create_time from mylock where id='1'").executeQuery();if (query.next()) {Date date = query.getDate(1);// 防止死锁，超过20秒删除if (date.getTime() + 20000L > System.currentTimeMillis()) {conn.prepareStatement("delete from mylock where id='1' and create_time='" + date.getTime() + "'").executeUpdate();// 等待一下ThreadUtil.sleep(1000);}}query.close();}// 等待一下ThreadUtil.sleep(200);}}
}

4、执行结果正确

在这里插入图片描述

Zookeeper

使用zookeeper有多钟方案，如下：

每个客户端在zookeeper的一个指定目录下创建一个临时节点，通过判断当前目录下的所有节点与自己的节点的顺序，就可以确定自己是否获取到锁。
每个客户端在zookeeper的一个指定目录下创建一个有序节点，通过判断自己节点在所有子节点中的顺序，就可以确定自己是否获取到锁。

第一种方案与数据库类似

Zookeeper 方案一

每个客户端在zookeeper的一个指定目录下创建一个临时节点，通过判断当前目录下的所有节点与自己的节点的顺序，就可以确定自己是否获取到锁。

1、Maven中添加依赖

<dependency><groupId>cn.hutool</groupId><artifactId>hutool-core</artifactId><version>5.8.27</version>
</dependency>
<dependency><groupId>org.apache.zookeeper</groupId><artifactId>zookeeper</artifactId><version>3.9.2</version>
</dependency>

2、找个Zookeeper启动好

3、java实现代码

package zk;import cn.hutool.core.io.IoUtil;
import cn.hutool.core.thread.ThreadUtil;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;/*** @author lingkang* @create by 2024/7/25 18:17*/
public class Demo01 {/*** 假设是库存*/private static int number = 100;private static final String lockPath = "/lock";public static void main(String[] args) throws Exception {// 提前初始化好redis连接List<ZooKeeper> zooKeeperList = new ArrayList<>();for (int i = 0; i < 2; i++) {ZooKeeper zooKeeper = new ZooKeeper("10.8.4.191:2181", 20000, null);// 用来做一个初始化调用zooKeeper.exists(lockPath, false);zooKeeperList.add(zooKeeper);}// 操作前先清空锁ZooKeeper zk = zooKeeperList.get(0);Stat exists = zk.exists(lockPath, false);if (exists != null) {zk.delete(lockPath, exists.getVersion());}// 假设有2个线程进行锁操作List<Thread> threadList = new ArrayList<>();for (int i = 0; i < 2; i++) {final ZooKeeper zooKeeper = zooKeeperList.get(i);Thread thread = new Thread(() -> {// 每隔线程减扣 5次for (int j = 0; j < 5; j++) {try {dd(zooKeeper);} catch (Exception e) {throw new RuntimeException(e);}}IoUtil.close(zooKeeper);// 注意此线程id，此线程id用于模拟全局唯一业务处理id，实际开发可以用具体业务id替代，必须全局唯一}, "t-" + i);thread.start();threadList.add(thread);}for (Thread thread : threadList)thread.join();ThreadUtil.sleep(1000);// 应该输出 100-5*2 --> 90System.out.println("库存: " + number);}/*** 业务处理*/private static void dd(ZooKeeper zooKeeper) throws Exception {String threadName = Thread.currentThread().getName();for (; ; ) {Stat stat = zooKeeper.exists(lockPath, false);if (stat != null) {byte[] data = zooKeeper.getData(lockPath, null, stat);if (!threadName.equals(new String(data))) {// 说明已经被其他服务获取锁了，等待一下跳过此次锁创建ThreadUtil.sleep(300);// 等待一下continue;}}try {// 创建者将会获得锁，注意：注意此线程id，此线程id用于模拟全局唯一业务处理id，// 实际开发可以用具体业务id替代，必须全局唯一zooKeeper.create(lockPath, threadName.getBytes(StandardCharsets.UTF_8), ZooDefs.Ids.CREATOR_ALL_ACL, CreateMode.EPHEMERAL);} catch (Exception e) {// 创建锁失败}// 等待一下ThreadUtil.sleep(200);stat = zooKeeper.exists(lockPath, false);// 如果是当前session创建的，就获得锁if (stat == null) {continue;} else {byte[] data = zooKeeper.getData(lockPath, null, stat);if (!threadName.equals(new String(data))) {// 说明已经被其他服务获取锁了，等待一下跳过此次锁创建ThreadUtil.sleep(200);// 等待一下continue;}}// 获取到锁System.out.println("当前线程获得锁：" + threadName);// 进行减扣库存等一系列操作....number--;// 假设处理业务延迟一下ThreadUtil.sleep(200);// 处理完毕要移除锁zooKeeper.delete(lockPath, stat.getVersion());// 处理完成break;}}
}

4、执行结果正确

在这里插入图片描述

Zookeeper 方案二（推荐）

每个客户端在zookeeper的一个指定目录下创建一个有序节点，通过判断自己节点在所有子节点中的顺序，就可以确定自己是否获取到锁。

1、Maven中添加依赖

<dependency><groupId>cn.hutool</groupId><artifactId>hutool-core</artifactId><version>5.8.27</version>
</dependency>
<!-- 由于我的zk版本是3.4.8 版本较低，所以使用低版本curator-recipes -->
<dependency><groupId>org.apache.curator</groupId><artifactId>curator-recipes</artifactId><version>2.13.0</version>
</dependency>

2、找个Zookeeper启动好

v3.4.8

3、java实现代码

package zk;import cn.hutool.core.io.IoUtil;
import cn.hutool.core.thread.ThreadUtil;
import org.apache.curator.RetryPolicy;
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 org.apache.zookeeper.data.Stat;import java.util.ArrayList;
import java.util.List;/*** @author lingkang* @create by 2024/7/26 10:43*/
public class Demo02 {/*** 假设是库存*/private static int number = 100;private static final String lockPath = "/lock";public static void main(String[] args) throws Exception {// 提前初始化好zk连接List<CuratorFramework> curatorFrameworks = new ArrayList<>();for (int i = 0; i < 2; i++) {RetryPolicy retryPolicy = new ExponentialBackoffRetry(2000, 3);CuratorFramework client = CuratorFrameworkFactory.newClient("10.8.4.191:2181", retryPolicy);client.start();curatorFrameworks.add(client);}// 操作前先清空锁CuratorFramework curator = curatorFrameworks.get(0);Stat stat = curator.checkExists().forPath(lockPath);if (stat != null)curator.delete().withVersion(stat.getVersion()).forPath(lockPath);// 假设有JVM进行锁操作List<Thread> threadList = new ArrayList<>();for (int i = 0; i < 2; i++) {final CuratorFramework curatorFramework = curatorFrameworks.get(i);Thread thread = new Thread(() -> {// 每个线程减扣 5次for (int j = 0; j < 5; j++) {try {dd(curatorFramework);} catch (Exception e) {throw new RuntimeException(e);}}}, "t-" + i);threadList.add(thread);thread.start();}for (Thread thread : threadList)thread.join();ThreadUtil.sleep(1000);for (CuratorFramework curatorFramework : curatorFrameworks)IoUtil.close(curatorFramework);System.out.println("-------------------------------------------------------------------");// 应该输出 100-5*2 --> 90System.out.println("库存: " + number);}/*** 业务处理*/private static void dd(CuratorFramework curatorFramework) {String threadName = Thread.currentThread().getName();InterProcessMutex lock = new InterProcessMutex(curatorFramework, lockPath);try {lock.acquire();// 获取到锁，执行业务逻辑System.out.println("当前线程获得锁：" + threadName);// 进行减扣库存等一系列操作....number--;// 假设处理业务延迟一下ThreadUtil.sleep(200);} catch (Exception e) {// 处理异常e.printStackTrace();} finally {try {lock.release();// 释放锁} catch (Exception e) {// 处理释放锁时的异常e.printStackTrace();}}}
}