今日皆さんと共有したいのは、分散ロックです。この記事では、5 つの事例、図、ソース コード分析などを使用します。分析します。
共通の同期ロック、およびその他のロックはすべて、単一の JVM に基づいて実装されます。分散シナリオではどうすればよいですか?このとき、分散ロックが登場しました。
分散実装ソリューションに関しては、業界で 3 つの人気のあるソリューションがあります。
1、データベースに基づく
2、Redis## に基づく
Zookeeper
etcd や consul を使用した実装もあります。
開発で最も一般的に使用される 2 つのソリューションは Redis と Zookeeper です。2 つのソリューションの中で最も複雑で、問題を引き起こす可能性が最も高いのは Redis## です。 # 実装計画ですので、今日は Redis 実装計画について説明します。
@RestController
public class IndexController {
@Autowired
private StringRedisTemplate stringRedisTemplate;
/**
* 模拟下单减库存的场景
* @return
*/
@RequestMapping(value = "/duduct_stock")
public String deductStock(){
// 从redis 中拿当前库存的值
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
if(stock > 0){
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock",realStock + "");
System.out.println("扣减成功，剩余库存：" + realStock);
}else{
System.out.println("扣减失败，库存不足");
}
return "end";
}
}Redis 値は 100 です。
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));这行代码，获取到的值都为100，紧跟着判断大于0后都进行-1操作，最后设置到redis 中的值都为99。但正常执行完成后redis中的值应为 95。
在遇到案例1的问题后，大部分人的第一反应都会想到加锁来控制事务的原子性，如下代码所示：
@RequestMapping(value = "/duduct_stock")
public String deductStock(){
synchronized (this){
// 从redis 中拿当前库存的值
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
if(stock > 0){
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock",realStock + "");
System.out.println("扣减成功，剩余库存：" + realStock);
}else{
System.out.println("扣减失败，库存不足");
}
}
return "end";
}现在当有多个请求访问该接口时，同一时刻只有一个请求可进入方法体中进行库存的扣减，其余请求等候。
但我们都知道，synchronized 锁是属于JVM级别的，也就是我们俗称的“单机锁”。但现在基本大部分公司使用的都是集群部署，现在我们思考下以上代码在集群部署的情况下还能保证库存数据的一致性吗？
答案是不能，如上图所示，请求经Nginx分发后，可能存在多个服务同时从Redis中获取库存数据，此时只加synchronized （单机锁）是无效的，并发越高，出现问题的几率就越大。
setnx：将 key 的值设为 value，当且仅当 key 不存在。
若给定 key 已经存在，则 setnx 不做任何动作。
使用setnx实现简单的分布式锁：
/**
* 模拟下单减库存的场景
* @return
*/
@RequestMapping(value = "/duduct_stock")
public String deductStock(){
String lockKey = "product_001";
// 使用 setnx 添加分布式锁
// 返回 true 代表之前redis中没有key为 lockKey 的值，并已进行成功设置
// 返回 false 代表之前redis中已经存在 lockKey 这个key了
Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "wangcp");
if(!result){
// 代表已经加锁了
return "error_code";
}
// 从redis 中拿当前库存的值
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
if(stock > 0){
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock",realStock + "");
System.out.println("扣减成功，剩余库存：" + realStock);
}else{
System.out.println("扣减失败，库存不足");
}
// 释放锁
stringRedisTemplate.delete(lockKey);
return "end";
}我们知道 Redis 是单线程执行，现在再看案例2中的流程图时，哪怕高并发场景下多个请求都执行到了setnx的代码，redis会根据请求的先后顺序进行排列，只有排列在队头的请求才能设置成功。其它请求只能返回“error_code”。
当setnx设置成功后，可执行业务代码对库存扣减，执行完成后对锁进行释放。
我们再来思考下以上代码已经完美实现分布式锁了吗？能够支撑高并发场景吗？答案并不是，上面的代码还是存在很多问题的，离真正的分布式锁还差的很远。
我们分析一下，上面的代码存在的问题：
死锁：假如第一个请求在setnx加锁完成后，执行业务代码时出现了异常，那释放锁的代码就无法执行，后面所有的请求也都无法进行操作了。
针对死锁的问题，我们对代码再次进行优化，添加try-finally，在finally中添加释放锁代码，这样无论如何都会执行释放锁代码，如下所示：
/**
* 模拟下单减库存的场景
* @return
*/
@RequestMapping(value = "/duduct_stock")
public String deductStock(){
String lockKey = "product_001";
try{
// 使用 setnx 添加分布式锁
// 返回 true 代表之前redis中没有key为 lockKey 的值，并已进行成功设置
// 返回 false 代表之前redis中已经存在 lockKey 这个key了
Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "wangcp");
if(!result){
// 代表已经加锁了
return "error_code";
}
// 从redis 中拿当前库存的值
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
if(stock > 0){
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock",realStock + "");
System.out.println("扣减成功，剩余库存：" + realStock);
}else{
System.out.println("扣减失败，库存不足");
}
}finally {
// 释放锁
stringRedisTemplate.delete(lockKey);
}
return "end";
}经过改进后的代码是否还存在问题呢？我们思考正常执行的情况下应该是没有问题，但我们假设请求在执行到业务代码时服务突然宕机了，或者正巧你的运维同事重新发版，粗暴的 kill -9 掉了呢，那代码还能执行 finally 吗？
针对想到的问题，对代码再次进行优化，加入过期时间，这样即便出现了上述的问题，在时间到期后锁也会自动释放掉，不会出现“死锁”的情况。
@RequestMapping(value = "/duduct_stock")
public String deductStock(){
String lockKey = "product_001";
try{
Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey,"wangcp",10,TimeUnit.SECONDS);
if(!result){
// 代表已经加锁了
return "error_code";
}
// 从redis 中拿当前库存的值
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
if(stock > 0){
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock",realStock + "");
System.out.println("扣减成功，剩余库存：" + realStock);
}else{
System.out.println("扣减失败，库存不足");
}
}finally {
// 释放锁
stringRedisTemplate.delete(lockKey);
}
return "end";
}现在我们再思考一下，给锁加入过期时间后就可以了吗？就可以完美运行不出问题了吗？
超时时间设置的10s真的合适吗？如果不合适设置多少秒合适呢？如下图所示
同時に 3 つのリクエストがあると仮定します。
3 つのリクエストをシミュレートするだけで問題がわかります。実際に同時実行性が高いシナリオの場合、ロックは「常に無効」または「永続的に無効」になる可能性があります。
それでは、具体的な問題はどこにあるのでしょうか?概要は次のとおりです:
問題に対する対応する解決策を検討します。
Spring Boot統合Redissonステップ
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.6.5</version>
</dependency>@Bean
public RedissonClient redisson(){
// 单机模式
Config config = new Config();
config.useSingleServer().setAddress("redis://192.168.3.170:6379").setDatabase(0);
return Redisson.create(config);
}@RestController
public class IndexController {
@Autowired
private RedissonClient redisson;
@Autowired
private StringRedisTemplate stringRedisTemplate;
/**
* 模拟下单减库存的场景
* @return
*/
@RequestMapping(value = "/duduct_stock")
public String deductStock(){
String lockKey = "product_001";
// 1.获取锁对象
RLock redissonLock = redisson.getLock(lockKey);
try{
// 2.加锁
redissonLock.lock(); // 等价于 setIfAbsent(lockKey,"wangcp",10,TimeUnit.SECONDS);
// 从redis 中拿当前库存的值
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
if(stock > 0){
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock",realStock + "");
System.out.println("扣减成功，剩余库存：" + realStock);
}else{
System.out.println("扣减失败，库存不足");
}
}finally {
// 3.释放锁
redissonLock.unlock();
}
return "end";
}
}
我们点击lock()方法，查看源码，最终看到以下代码
<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
internalLockLeaseTime = unit.toMillis(leaseTime);
return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
"if (redis.call('exists', KEYS[1]) == 0) then " +
"redis.call('hset', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
"return redis.call('pttl', KEYS[1]);",
Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
}没错，加锁最终执行的就是这段lua 脚本语言。
if (redis.call('exists', KEYS[1]) == 0) then
redis.call('hset', KEYS[1], ARGV[2], 1);
redis.call('pexpire', KEYS[1], ARGV[1]);
return nil;
end;脚本的主要逻辑为：
这样来看其实和我们前面案例中的实现方法好像没什么区别，但实际上并不是。
这段lua脚本命令在Redis中执行时，会被当成一条命令来执行，能够保证原子性，故要不都成功，要不都失败。
我们在源码中看到Redssion的许多方法实现中很多都用到了lua脚本，这样能够极大的保证命令执行的原子性。
下面是Redisson锁自动“续命”源码：
private void scheduleExpirationRenewal(final long threadId) {
if (expirationRenewalMap.containsKey(getEntryName())) {
return;
}
Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
@Override
public void run(Timeout timeout) throws Exception {
RFuture<Boolean> future = commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return 1; " +
"end; " +
"return 0;",
Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
future.addListener(new FutureListener<Boolean>() {
@Override
public void operationComplete(Future<Boolean> future) throws Exception {
expirationRenewalMap.remove(getEntryName());
if (!future.isSuccess()) {
log.error("Can't update lock " + getName() + " expiration", future.cause());
return;
}
if (future.getNow()) {
// reschedule itself
scheduleExpirationRenewal(threadId);
}
}
});
}
}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
if (expirationRenewalMap.putIfAbsent(getEntryName(), task) != null) {
task.cancel();
}
}这段代码是在加锁后开启一个守护线程进行监听。Redisson超时时间默认设置30s，线程每10s调用一次判断锁还是否存在，如果存在则延长锁的超时时间。
现在，我们再回过头来看看案例5中的加锁代码与原理图，其实完善到这种程度已经可以满足很多公司的使用了，并且很多公司也确实是这样用的。但我们再思考下是否还存在问题呢？例如以下场景：
Redis は、実際のデプロイと使用ではクラスターにデプロイされます。同時実行性の高いシナリオでは、ロックされます。マスター ノードにキーを書き込んだ後、マスターまだスレーブノードと同期していないときにマスターがダウンし、元のスレーブノードが選出後に新しいマスターノードになりましたが、このときロック失敗の問題が発生する可能性があります。 Redis はリクエストの実行をキューに入れるのに役立ちます。これは、並列処理をシリアル化に変換することを意味します。シリアルに実行されるコードには同時実行性の問題はまったくありませんが、プログラムのパフォーマンスは確実に影響を受けます。 考え方 問題を解決するときは、まず CAP 原則 (一貫性、可用性、パーティション耐性) を考え、次に現在の Redis が AP (可用性、パーティション耐性) を満たすことを考えます。 CP (一貫性、パーティションフォールトトレランス) を満たす分散システムを見つける必要があります。最初に思い浮かぶのは Zookeeper です。Zookeeper のクラスター間データ同期メカニズムは、マスター ノードがデータを受信したときに、マスター ノードに成功したフィードバックをすぐに返さないことです。クライアントは最初に子ノードと通信し、同期を行うと、半分以上のノードが同期を完了した後でのみクライアントに受信成功の通知が届きます。また、マスター ノードがダウンした場合、Zookeeper の Zab プロトコル (Zookeeper アトミック ブロードキャスト) に従って再選出されたマスター ノードが正常に同期されている必要があります。
そこで問題は、Redisson と Zookeeper の分散ロックのどちらをどのように選択するかということです。答えは、同時実行の量がそれほど高くない場合、Zookeeper を使用して分散ロックを実行できますが、その同時実行機能は Redis よりもはるかに劣ります。比較的高い同時実行要件がある場合は、Redis を使用してください。時折発生するマスター/スレーブ アーキテクチャのロック障害の問題は、実際には許容できる程度です。
パフォーマンス向上の 2 番目の問題に関しては、ConcurrentHashMap のロック セグメンテーション テクノロジのアイデアを参照できます。コードのインベントリ 現在の量は 1000 ですが、それを 10 個のセグメントに分割し、各セグメントを 100 にして、各セグメントを個別にロックすることで、10 個のリクエストのロックと処理を同時に実行できます。 、要件がある学生は引き続き細分化できます。しかし実際には、Redis の Qps は 10W に達しており、特に高い同時実行性がないシナリオでは十分です。
以上が分散錠：5件 玄関から埋葬までの詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。
![[Web フロントエンド] Node.js クイック スタート](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
