PHP多并发加锁技巧_并发情况若何实现加锁来担保数据一致性

文章目录 [+]

大略的与我们常用的 Synchronized 进行比较：

ReentrantLock

PHP多并发加锁技巧_并发情况若何实现加锁来担保数据一致性

Synchronized

（图片来自网络侵删）

锁实现机制

依赖 AQS

监视器模式

灵巧性

支持相应超时、中断、考试测验获取锁

不灵巧

开释形式

必须显示调用 unlock () 开释锁

自动开释监视器

锁类型

公正锁 & 非公正锁

非公正锁

条件行列步队

可关联多个条件行列步队

关联一个条件行列步队

可重入性

可重入

AQS 机制：如果被要求的共享资源空闲，那么就当前要求资源的线程设置为有效的事情线程，将共享资源通过 CAScompareAndSetState设置为锁定状态；如果共享资源被占用，就采取一定的壅塞等待唤醒机制（CLH 变体的 FIFO 双端行列步队）来担保锁分配。

可重入性：无论是公正锁还是非公正锁的情形，加锁过程会利用一个 state 值

private volatile int statestate 值初始化的时候为 0，表示没有任何线程持有锁当有线程来要求该锁时，state 值会自增 1，同一个线程多次获取锁，就会多次 + 1，这便是可重入的观点解锁也是对 state 值自减 1，一贯到 0，此线程对锁开释。

public class LockExample { static int count = 0; static ReentrantLock lock = new ReentrantLock(); public static void main(String[] args) throws InterruptedException { Runnable runnable = new Runnable() { @Override public void run() { try { // 加锁 lock.lock(); for (int i = 0; i < 10000; i++) { count++; } } catch (Exception e) { e.printStackTrace(); } finally { // 解锁，放在finally子句中，担保锁的开释 lock.unlock(); } } }; Thread thread1 = new Thread(runnable); Thread thread2 = new Thread(runnable); thread1.start(); thread2.start(); thread1.join(); thread2.join(); System.out.println("count: " + count); }}/ 输出 count: 20000 /2. Mysql 行锁、乐不雅观锁

乐不雅观锁即是无锁思想，一样平常都是基于 CAS 思想实现的，而在 MySQL 中通过 version 版本号 + CAS 无锁形式实现乐不雅观锁；例如 T1，T2 两个事务一起并发实行时，当 T2 事务实行成功提交后，会对 version+1，以是 T1 事务实行的 version 条件就无法成立了。

对 sql 语句进行加锁以及状态机的操作，也可以避免不同线程同时对 count 值访问导致的数据不一致问题。

// 乐不雅观锁 + 状态机update table_nameset version = version + 1, count = count + 1where id = id AND version = version AND count = [修正前的count值];// 行锁 + 状态机 update table_nameset count = count + 1where id = id AND count = [修正前的count值]for update;3. 细粒度的 ReetrantLock 锁

如果我们直接采取 ReentrantLock 全局加锁，那么这种情形是一条线程获取到锁，全体程序全部的线程来到这里都会壅塞；但是我们在项目里面想要针对每个用户在操作的时候实现互斥逻辑，以是我们须要更加细粒度的锁。

public class LockExample { private static Map<String, Lock> lockMap = new ConcurrentHashMap<>(); public static void lock(String userId) { // Map中添加细粒度的锁资源 lockMap.putIfAbsent(userId, new ReentrantLock()); // 从容器中拿锁并实现加锁 lockMap.get(userId).lock(); } public static void unlock(String userId) { // 先从容器中拿锁，确保锁的存在 Lock locak = lockMap.get(userId); // 开释锁 lock.unlock(); }}

弊端：如果每一个用户要求共享资源，就会加锁一次，后续该用户就没有在登录过平台，但是锁工具会一贯存在于内存中，这等价于发生了内存泄露，以是锁的超时和淘汰机制机制须要实现。

4. 细粒度的 Synchronized 全局锁

上面的加锁机制利用到了锁容器ConcurrentHashMap，该随意马虎为了线程安全的情形，多以底层还是会用到Synchronized机制，以是有些情形，利用 lockMap 须要加上两层锁。

那么我们是不是可以直策应用Synchronized来实现细粒度的锁机制

public class LockExample { public static void syncFunc1(Long accountId) { String lock = new String(accountId + "").intern(); synchronized (lock) { System.out.println(Thread.currentThread().getName() + "拿到锁了"); // 仿照业务耗时 try { Thread.sleep(1000); } catch (InterruptedException e) { throw new RuntimeException(e); } System.out.println(Thread.currentThread().getName() + "开释锁了"); } } public static void syncFunc2(Long accountId) { String lock = new String(accountId + "").intern(); synchronized (lock) { System.out.println(Thread.currentThread().getName() + "拿到锁了"); // 仿照业务耗时 try { Thread.sleep(1000); } catch (InterruptedException e) { throw new RuntimeException(e); } System.out.println(Thread.currentThread().getName() + "开释锁了"); } } // 利用 Synchronized 来实现更加细粒度的锁 public static void main(String[] args) { new Thread(()-> syncFunc1(123456L), "Thread-1").start(); new Thread(()-> syncFunc2(123456L), "Thread-2").start(); }}/ 打印 Thread-1拿到锁了 Thread-1开释锁了 Thread-2拿到锁了 Thread-2开释锁了 /从代码中我们创造实现加锁的工具实在便是一个与用户 ID 干系的一个字符串工具，这里可能会有疑问，我每一个新的线程进来，new 的都是一个新的字符串工具，只不过字符串内容一样，怎么能够担保可以安全的锁住共享资源呢；这实在须要归功于后面的intern()函数的功能；intern()函数用于在运行时将字符串添加到堆空间中的字符串常量池中，如果字符串已经存在，返回字符串常量池中的引用。
分布式架构下锁的实现方案

核心问题：我们须要找到一个多个进程之间所有线程可见的区域来定义这个互斥量。

一个精良的分布式锁的实现方案该当知足如下几个特性：

分布式环境下，可以担保不同进程之间的线程互斥同一时候，同时只许可一条线程成功获取到锁资源担保互斥量的地方须要担保高可用性要担保可以高性能的获取锁和开释锁可以支持同一线程的锁重入性具备合理的壅塞机制，竞争锁失落败的线程要有相应的处理方案支持非壅塞式的获取锁。
获取锁失落败的线程可以直接返回具备合理的锁失落效机制，如超时失落效等，可以确保避免去世锁情形涌现Redis 实现分布式锁redis 属于中间件，可独立支配；对付不同的 Java 进程来说都是可见的，同时性能也非常可不雅观依赖与 redis 本身供应的指令setnx key value来实现分布式锁；差异于普通set指令的是只有当 key 不存在时才会设置成功，key 存在时会返回设置失落败

代码实例：

// 扣库存接口@RequestMapping("/minusInventory")public String minusInventory(Inventory inventory) { // 获取锁 String lockKey = "lock-" + inventory.getInventoryId(); int timeOut = 100; Boolean flag = stringRedisTemplate.opsForValue() .setIfAbsent(lockKey, "竹子-熊猫",timeOut,TimeUnit.SECONDS); // 加上过期韶光，可以担保去世锁也会在一定韶光内开释锁 stringRedisTemplate.expire(lockKey,timeOut,TimeUnit.SECONDS); if(!flag){ // 非壅塞式实现 return "做事器繁忙...请稍后重试！！！ "; } // ----只有获取锁成功才能实行下述的减库存业务---- try{ // 查询库存信息 Inventory inventoryResult = inventoryService.selectByPrimaryKey(inventory.getInventoryId()); if (inventoryResult.getShopCount() <= 0) { return "库存不敷，请联系卖家...."; } // 扣减库存 inventoryResult.setShopCount(inventoryResult.getShopCount() - 1); int n = inventoryService.updateByPrimaryKeySelective(inventoryResult); } catch (Exception e) { // 确保业务涌现非常也可以开释锁，避免去世锁 // 开释锁 stringRedisTemplate.delete(lockKey); } if (n > 0) return "端口-" + port + ",库存扣减成功！！！ "; return "端口-" + port + ",库存扣减失落败！！！ ";}作者：竹子爱熊猫链接：https://juejin.cn/post/7038473714970656775

过期韶光的合理性剖析：

由于对付不同的业务，我们设置的过期韶光的是非都会不一样，太长了不得当，太短了也不得当；

以是我们想到的办理方案是设置一条子线程，给当前锁资源续命。
详细实现是，子线程间隔 2-3s 去查询一次 key 是否过期，如果还没有过期则代表业务线程还在实行业务，那么则为该 key 的过期韶光加上 5s。

但是为了避免主线程意外去世亡后，子线程会一贯为其续命，造成 “永生锁” 的征象，以是将子线程变为主（业务）线程的守护线程，这样子线程就会随着主线程一起去世亡。

// 续命子线程public class GuardThread extends Thread { private static boolean flag = true; public GuardThread(String lockKey, int timeOut, StringRedisTemplate stringRedisTemplate){ …… } @Override public void run() { // 开启循环续命 while (flag){ try { // 先休眠一半的韶光 Thread.sleep(timeOut / 2 1000); }catch (Exception e){ e.printStackTrace(); } // 韶光过了一半之后再去续命 // 先查看key是否过期 Long expire = stringRedisTemplate.getExpire( lockKey, TimeUnit.SECONDS); // 如果过期了，代表主线程开释了锁 if (expire <= 0){ // 停滞循环 flag = false; } // 如果还未过期 // 再为则续命一半的韶光 stringRedisTemplate.expire(lockKey,expire + timeOut/2,TimeUnit.SECONDS); } }}// 创建子线程为锁续命GuardThread guardThread = new GuardThread(lockKey,timeOut,stringRedisTemplate);// 设置为当前业务线程的守护线程guardThread.setDaemon(true);guardThread.start();作者：竹子爱熊猫链接：https://juejin.cn/post/7038473714970656775Redis 主从架构下锁失落效的问题

为了在开拓过程担保 Redis 的高可用，会采取主从复制架构做读写分离，从而提升 Redis 的吞吐量以及可用性。
但是如果一条线程在 redis 主节点上获取锁成功之后，主节点还没有来得及复制给从节点就宕机了，此时另一条线程访问 redis 就会在从节点上面访问，同时也获取锁成功，这时候临界资源的访问就会涌现安全性问题了。

办理办法：

红锁算法（官方提出的办理方案）：多台独立的 Redis 同时写入数据，在锁失落效韶光之内，一半以上的机器写成功则返回获取锁成功，失落败的时候开释掉那些成功的机器上的锁。
但这种做法缺陷是本钱高须要独立支配多台 Redis 节点。
额外记录锁状态：再额外通过其他独立支配的中间件（比如 DB）来记录锁状态，在新线程获取锁之前须要先查询 DB 中的锁持有记录，只要当锁状态为未持有时再考试测验获取分布式锁。
但是这种情形缺陷显而易见，获取锁的过程实现难度繁芜，性能开销也非常大；其余还须要合营定时器功能更新 DB 中的锁状态，担保锁的合理失落效机制。
利用 Zookepper 实现Zookeeper 实现分布式锁

Zookeeper 数据差异于 redis 的数据，数据是实时同步的，主节点写入后须要一半以上的节点都写入才会返回成功。
以是如果像电商、教诲等类型的项目追求高性能，可以放弃一定的稳定性，推举利用 redis 实现；例如像金融、银行、政府等类型的项目，追求高稳定性，可以捐躯一部分性能，推举利用 Zookeeper 实现。

分布式锁性能优化

上面加锁确实办理了并发情形下线程安全的问题，但是我们面对 100w 个用户同时去抢购 1000 个商品的场景该如何办理呢？

可与将共享资源做一下提前预热，分段分散存储一份。
抢购韶光为下午 15：00，提前再 14：30 旁边将商品数量分成 10 份，并将每一块数据进行分别加锁，来防止并发非常。
其余也须要在 redis 中写入 10 个 key，每一个新的线程进来先随机的分配一把锁，然后进行后面的减库存逻辑，完成之后开释锁，以便之后的线程利用。
这种分布式锁的思想便是，将原来一把锁就可以实现的多线程同步访问共享资源的功能，为了提高瞬时情形下多线程的访问速率，还须要担保并发安全的情形下一种实现办法。
参考文章：

1. https://juejin.cn/post/7236213437800890423

2. https://juejin.cn/post/7038473714970656775

3. https://tech.meituan.com/2019/12/05/aqs-theory-and-apply.html

作者：京东科技焦泽斌

来源：京东云开拓者社区转载请注明来源