前言一、基于数据库实现分布式锁二、Redisson实现分布式锁三、Zookeeper实现分布式锁四、总结
前言
目前几乎很多大型网站及应用都是分布式部署的，分布式场景中的数据一致性问题一直是一个比较重要的话题。分布式的CAP理论告诉我们“任何一个分布式系统都无法同时满足一致性（Consistency）、可用性（Availability）和分区容错性（Partition tolerance），最多只能同时满足两项。”所以，很多系统在设计之初就要对这三者做出取舍。在互联网领域的绝大多数的场景中，都需要牺牲强一致性来换取系统的高可用性，系统往往只需要保证“最终一致性”，只要这个最终时间是在用户可以接受的范围内即可。
在很多场景中，我们为了保证数据的最终一致性，需要很多的技术方案来支持，比如分布式事务、分布式锁等。有的时候，我们需要保证一个方法在同一时间内只能被同一个线程执行。
分布式锁有三种实现方式：
1.基于数据库实现分布式锁；
2.基于缓存（Redis等）实现分布式锁；
3.基于Zookeeper实现分布式锁；
本篇只是基于单JVM做测试。
一、基于数据库实现分布式锁
- 建表语句
其中lock_key为锁的名称，并设置唯一索引。
CREATE TABLE `distribute_lock` ( `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT, `lock_key` varchar(50) COLLATE utf8mb4_unicode_ci DEFAULT NULL, PRIMARY KEY (`id`), UNIQUE KEY `lock_key` (`lock_key`) ) ENGINE=InnoDB AUTO_INCREMENT=8 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
- 实体类
import com.baomidou.mybatisplus.annotation.*;
import lombok.Data;
import java.io.Serializable;
@TableName("distribute_lock")
@Data
public class DistributeLock implements Serializable {
@TableId(value = "id", type = IdType.AUTO)
private Long id;
@TableField("lock_key")
private String lockKey;
}
3. 测试
此处数据库相关操作使用Mybatis-Plus实现。
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.entity.DistributeLock;
import com.service.LockService;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
@RunWith(SpringRunner.class)
@SpringBootTest
@Slf4j
public class MySQLDistributedLock {
@Autowired
private LockService lockService;
//锁名称
private static String LOCK_KEY = "lock_key";
@Test
public void testLock() throws InterruptedException {
new Thread(new Runnable() {
@Override
public void run() {
boolean flag = false;
try {
DistributeLock lock = new DistributeLock();
lock.setLockKey(LOCK_KEY);
flag = lockService.save(lock);
if (flag) {
log.info(Thread.currentThread().getName() + ":线程1获取锁成功");
//模拟业务场景
Thread.sleep(3000);
}
} catch (Exception e) {
log.error(Thread.currentThread().getName() + ":线程1获取锁异常");
} finally {
//只有抢锁成功才能释放锁,防止将其他线程持有的锁释放
if (flag) {
lockService.remove(Wrappers.lambdaQuery().eq(DistributeLock::getLockKey, LOCK_KEY));
log.info(Thread.currentThread().getName() + ":线程1释放锁成功");
}
}
}
}).start();
new Thread(new Runnable() {
@Override
public void run() {
boolean flag = false;
try {
DistributeLock lock = new DistributeLock();
lock.setLockKey(LOCK_KEY);
flag = lockService.save(lock);
if (flag) {
log.info(Thread.currentThread().getName() + ":线程2获取锁成功");
//模拟业务场景
Thread.sleep(3000);
}
} catch (Exception e) {
log.error(Thread.currentThread().getName() + ":线程2获取锁异常");
} finally {
//只有抢锁成功才能释放锁,防止将其他线程持有的锁释放
if (flag) {
lockService.remove(Wrappers.lambdaQuery().eq(DistributeLock::getLockKey, LOCK_KEY));
log.info(Thread.currentThread().getName() + ":线程2释放锁成功");
}
}
}
}).start();
//睡眠10秒钟,防止主线程执行完关闭程序,而其他线程还未执行
Thread.sleep(10000);
}
}
4. 输出结果
结果显示线程2先抢到锁，即先插入一条数据到数据库中，此时线程1再去插表时因为唯一索引导致插表失败,即抢锁失败，这种方式实现分布式锁时不可重入的。
1. 依赖
org.redisson redisson 3.8.2
2. 实体类
import com.baomidou.mybatisplus.annotation.*;
import lombok.Data;
import java.io.Serializable;
@TableName("distribute_lock")
@Data
public class DistributeLock implements Serializable {
@TableId(value = "id", type = IdType.AUTO)
private Long id;
@TableField("lock_key")
private String lockKey;
}
3. 测试
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 RedisDistributedLock {
public static void main(String[] args) throws InterruptedException {
//单机模式下构造redisson必须的配置
Config config = new Config();
config.useSingleServer().setAddress("redis://127.0.0.1:6379").setDatabase(0);
RedissonClient redissonClient = Redisson.create(config);
new Thread(new Runnable() {
@Override
public void run() {
//此处多线程获取锁的名称必须相同,代表获取同一把锁
RLock lock = redissonClient.getLock("lock");
try {
if (lock.tryLock(3, 10, TimeUnit.SECONDS)) {
System.out.println(Thread.currentThread().getName() + ":线程1第一次获取锁成功");
//实现可重入特性
if (lock.tryLock(3, 5, TimeUnit.SECONDS)) {
System.out.println(Thread.currentThread().getName() + ":线程1第二次获取锁成功");
}
//此处睡眠10秒演示线程2抢锁超时失败
Thread.sleep(10000);
} else {
System.out.println(Thread.currentThread().getName() + ":线程1获取锁失败");
}
} catch (InterruptedException e) {
System.out.println(Thread.currentThread().getName() + ":线程1获取锁异常");
} finally {
lock.unlock();
}
}
}).start();
new Thread(new Runnable() {
@Override
public void run() {
RLock lock = null;
try {
//此处睡眠1秒为防止线程2抢先执行抢锁
Thread.sleep(1000);
lock = redissonClient.getLock("lock");
if (lock.tryLock(3, 5, TimeUnit.SECONDS)) {
System.out.println(Thread.currentThread().getName() + ":线程2获取锁成功");
} else {
System.out.println(Thread.currentThread().getName() + ":线程2获取锁失败");
}
} catch (InterruptedException e) {
System.out.println(Thread.currentThread().getName() + ":线程2获取锁异常");
} finally {
lock.unlock();
}
}
}).start();
}
}
4. 输出结果
线程1两次获取锁成功，实现了锁的可重入特性，线程2由于线程1抢到锁后睡眠10秒钟，导致线程2抢锁超时失败。
此处将线程1持锁时间改为5秒，线程2抢锁时间改为8秒，再次运行。
线程1修改代码：
//修改线程1持锁时间为5秒
if (lock.tryLock(3, 5, TimeUnit.SECONDS)) {
System.out.println(Thread.currentThread().getName() + ":线程1第一次获取锁成功");
//实现可重入特性
if (lock.tryLock(3, 5, TimeUnit.SECONDS)) {
System.out.println(Thread.currentThread().getName() + ":线程1第二次获取锁成功");
}
//此处睡眠10秒演示线程2抢锁超时失败
Thread.sleep(10000);
} else {
System.out.println(Thread.currentThread().getName() + ":线程1获取锁失败");
}
线程1修改代码：
if (lock.tryLock(8, 5, TimeUnit.SECONDS)) {
System.out.println(Thread.currentThread().getName() + ":线程2获取锁成功");
} else {
System.out.println(Thread.currentThread().getName() + ":线程2获取锁失败");
}
输出结果：
此处即使线程1抢到锁后睡眠了10秒，但在超过持锁时间后还时会自动释放锁，所以线程2可以抢锁成功，此方法可以预防多线程情况下某个服务器的线程获取到锁后宕机造成死锁现象。
三、Zookeeper实现分布式锁
此处使用Curator框架来实现。
1. 测试
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.RetryOneTime;
import org.junit.Test;
import java.util.concurrent.TimeUnit;
public class ZookeeperDistributedLock {
private static final String ConNECTION = "127.0.0.1";
private static final String LOCK_PATH = "/lockPath";
@Test
public void zkLock() throws InterruptedException {
Curatorframework client = CuratorframeworkFactory.builder().connectString(CONNECTION)
.sessionTimeoutMs(60000)
.connectionTimeoutMs(15000)
.retryPolicy(new RetryOneTime(1000)).build();
//启动客户端
client.start();
//获取互斥锁
InterProcessMutex mutex = new InterProcessMutex(client, LOCK_PATH);
new Thread(() -> {
try {
//获取锁的最大时长为3秒
if (mutex.acquire(3, TimeUnit.SECONDS)) {
System.out.println(Thread.currentThread().getName() + ":线程1第一次获取锁成功");
//实现锁的可重入特性
if (mutex.acquire(3, TimeUnit.SECONDS)) {
System.out.println(Thread.currentThread().getName() + ":线程1第二次获取锁成功");
// Thread.sleep(10000);
mutex.release();
System.out.println(Thread.currentThread().getName() + ":线程1第一次释放锁成功");
} else {
System.out.println(Thread.currentThread().getName() + ":线程1第二次获取锁失败");
}
} else {
System.out.println(Thread.currentThread().getName() + ":线程1第一次获取锁失败");
}
} catch (Exception e) {
System.out.println(Thread.currentThread().getName() + ":线程1操作锁异常");
e.printStackTrace();
} finally {
try {
mutex.release();
System.out.println(Thread.currentThread().getName() + ":线程1第二次释放锁成功");
} catch (Exception e) {
System.out.println(Thread.currentThread().getName() + ":线程1释放锁异常");
e.printStackTrace();
}
}
}).start();
new Thread(() -> {
try {
//防止线程2抢先执行
Thread.sleep(1000);
if (mutex.acquire(5, TimeUnit.SECONDS)) {
System.out.println(Thread.currentThread().getName() + ":线程2获取锁成功");
} else {
System.out.println(Thread.currentThread().getName() + ":线程2获取锁失败");
}
} catch (Exception e) {
System.out.println(Thread.currentThread().getName() + ":线程2获取锁异常");
e.printStackTrace();
} finally {
try {
System.out.println(Thread.currentThread().getName() + ":线程2释放锁");
mutex.release();
} catch (Exception e) {
System.out.println(Thread.currentThread().getName() + ":线程2释放锁异常");
e.printStackTrace();
}
}
}).start();
//防止主线程执行完毕关闭程序
Thread.sleep(Integer.MAX_VALUE);
}
}
2. 输出结果
线程1可以两次获取锁，实现了锁的可重入性
此时放开线程1第二次获取锁时睡眠10秒的代码，再次运行结果：
线程2获取锁时，由于线程1没有释放锁，导致线程2抢锁失败，释放锁异常是因为没有抢到锁。
在目前分布式锁实现方案中，比较成熟、主流的方案有两种：
（1）基于Redis的分布式锁
（2）基于ZooKeeper的分布式锁
两种锁，分别适用的场景为：
（1）基于ZooKeeper的分布式锁，适用于对数据一致性要求较高，而并发量不是太大的场景；
（2）基于Redis的分布式锁，适用于并发量很大、性能要求很高的、而对数据一致性的要求较低，允许实现最终一致性的场景。
欢迎大家共同探讨，如有错误，还望指正！
参考资料
[1]什么是分布式锁？实现分布式锁的三种方式
[2]Zookeeper 分布式锁 - 图解 - 秒懂
