网站集群怎么做,百度关键词怎么优化,网站建设包括的内容有什么,中山精品网站建设渠道前言 分布式锁相信大家都有用过#xff0c;常见的分布式锁实现方式例如redis、zookeeper、数据库都可以实现#xff0c;而我们代码中强引用这些分布式锁的代码#xff0c;那么当我们以后想替换分布式锁的实现方式时#xff0c;需要修改代码的成本会很高#xff0c;于是我…前言 分布式锁相信大家都有用过常见的分布式锁实现方式例如redis、zookeeper、数据库都可以实现而我们代码中强引用这些分布式锁的代码那么当我们以后想替换分布式锁的实现方式时需要修改代码的成本会很高于是我们需要借鉴一些设计模式思想来设计下面我介绍下这三个分布式锁的实现逻辑以及我们项目中是怎么实现 实现方式 数据库实现 首先我们设计一张这样的表 CREATE TABLE lock (key varchar(128) C NOT NULL,created timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,version int(8) DEFAULT 1,PRIMARY KEY (key) ) ENGINEInnoDB DEFAULT CHARSETutf8 COMMENT 分布式锁表;首先方法开启事务当我们需要对某块业务上锁时确定加锁的颗粒度设置key的值执行插入语句这里我们可以设计的全面一点对比其他的分布式锁实现方式貌似还缺了点什么比如这个锁不可重入无法设置超时时间 其实这些我们也可以解决 首先可重入方面version字段记录重入的次数 insert into table (key) values (#key#) on duplicate key version total version 1key是唯一索引执行完这个语句后会在这一行添加行锁。这时候后续有两种可能性如果是当前线程A重复插入锁那么会更新version字段如果是其他线程B想要插入这个锁那么由于A持有的行锁还未释放所以B会阻塞 超时方面我们可以设置数据库的innodb_lock_wait_timeout参数来设置超时时间默认50s等待时间超时这个时间则会报错 由于insert语句在RR隔离级别会生成间隙锁在并发较高的情况下会产生死锁的情况所以建议在RC情况下使用 Redis实现 如果我们自己设计redis实现加锁的话我们第一个想到的就是setNx语法它的作用就是当key不存在的情况下将key 的值设置为 value同时返回1如果key已存在则不设置value并且返回0 如果还要加上超时时间那么还需要执行expire语法 setnx lock true expire lock 10使用这两个语法会产生一个问题就是这两个语法不是原子性的当执行一个语法后系统报错了那么超时时间就无法设置这样子这个key就无法过期 基于这个问题Redis官方将这两个指令组合在了一起解决Redis分布式锁原子性操作的问题 SET key value [EX 过期时间] NX 将key 的值设置为 value同时返回1如果key已存在则不设置value并且返回0同时加上超时时间这是一个原子性操作 由此可见我们自己实现需要踩很多坑市面上有成熟的redis实现的分布式锁框架redission我们直接用就可以了它帮我们把坑都踩了一遍不用重复造轮子了简单讲解下它的原理 首先讲一下它的键值kv结构 key为锁的keyvalue为一个hash对象 {key:线程idvalue:重入次数} value为什么这么设计我们要解决两个问题 1、删除锁时A线程把B线程持有的锁给删除了为了解决这个问题我们需要在锁中记录线程id删除时就可以判断锁是否是当前线程持有的一致才可以删除 2、实现可重入的逻辑所以需要在锁中记录重入次数每次重入次数1 不同于上面SET key value [EX 过期时间] NX方式redission的加锁逻辑是通过一段lua脚本来实现的redis的lua脚本可以实现原子性的操作下图是lua脚本 解锁逻辑也是一样的由于需要判断当前线程才能执行删除所以也需要通过lua脚本来实现删除的逻辑 当我们设置了过期时间后如果我们的业务执行时间超过了设定的过期时间那么锁会提前删除就会出现各种各样的问题。所以redission实现了一个watch dog逻辑。它是一个后台线程每10s检查一次将锁的过期时间延长 redis我们通过都会设置高可用最常见的方案就是主从或者哨兵但是它保证了高可用的同时无法保证高一致性。这样子当redis的主节点挂了从节点还没有同步到主节点的数据就变成了主节点那么锁就会发生丢失 redission提供了RedLock算法通过使用多个Redis实例各个实例之间没有主从关系相互独立超过一半节点加锁成功才算获取到锁。不过这种算法也不是一个完美的算法多个实例加锁效率低同时也会衍生出一些其他问题 Zookeeper实现 Zookeeper有很多种节点种类其中有一种节点种类叫做临时顺序节点这个节点有两个特性首先是当客户端向Zookeeper添加了这个节点后如果之后客户端挂了那么这个临时节点会被删除不会一直存在其次是这个节点是有序递增的。这些特性很适合用来做分布式锁 加锁就是在Zookeeper上添加临时顺序节点判断是否是最小节点如果是最小节点则无需排队直接执行。如果不是最小的则往后加一个顺序节点并且向前一个节点添加一个watch监听线程阻塞等待排队 当前一个节点删除时当前节点监听到删除事件并唤醒线程。这样子第一个通知第二个第二个通知第三个这种击鼓传花的方式可以避免羊群效应 羊群效应就是前一个节点释放锁后所有节点被唤醒这样会给服务器带来巨大压力 哪种更好 这个问题我觉得得根据我们的现实情况做判定当我们的系统只有数据库又不想依赖其他的中间件那我们使用数据库实现的方方式就可以了但是性能会很差容易出现瓶颈 Redis和Zookeeper性能都比数据库好这两者相比较而言Redis作为分布式锁大家使用的会多一些主要原因我想应该是Zookeeper的cp特性导致单leader节点易出现瓶颈而redis如果出现瓶颈后弹性伸缩增加节点会很方便所以性能更高 踩坑点 之前我们在代码中加入分布式锁时碰到过一个坑就是明明加入了分布式锁但是没有锁住代码。这里有几种原因我就不一一展开了我们之前踩过的一个坑就是我们的方法加了事务Transactional注解同时方法里面的逻辑加了redis分布式锁类似下面的代码。 Transactional private void test() {// 加锁// 查询 id 1的记录// 更新 id1的记录// 释放锁 }这时候当线程A进入了方法首先进入事务加锁然后执行方法里面的逻辑释放锁但是事务还未释放。这时候线程B也进入了这个方法锁因为已经释放了就直接进入方法逻辑了但是线程A的事务此时还没有提交所以线程B查询的id1的记录是不对的。 这个是踩坑了解决方法有两种一种是把事务的方法放在加锁的逻辑里面另外一种就是释放锁的逻辑改成监听spring 事务提交的事件实现事务完成后再释放锁我们最后也是这么改的 Transactional private void test() {// 加锁// 查询 id 1的记录// 更新 id1的记录// unlockAfterTransaction方法 }private void unlockAfterTransaction(LockResult lockResult) {//事物完成后释放锁TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {Overridepublic void afterCompletion(int status) {super.afterCompletion(status);distLockSservice.unlock(lockResult);}}); } 实战 这里再重审下为什么使用模板模式因为我们使用分布式锁的实现有多种如果我们在业务代码中直接依赖某一种分布式锁的话那么后续我们想替换分布式锁的实现会很麻烦所有有依赖的类都得替换所以我们使用模版模式把分布式锁的实现以及加锁、释放锁的逻辑放到公共代码中我们业务类只需要实现自己的业务逻辑即可无需关心分布式锁的相关逻辑调用方法即可 我们在1.8之前使用模版模式会比较繁琐我们需要准备一个抽象类定义一些公共的逻辑然后子类继承这个抽象类来自定义不同的逻辑。下面我以redission分布式锁的实战为例 public abstract class AbstractLockService {RedissonClient redissonClient;public void lock(ListString keyNames, Long timeout,Object... o) throws InterruptedException {// 加锁RLock[] locks new RLock[keyNames.size()];for(int i 0; i keyNames.size(); i) {locks[i] this.redissonClient.getLock(keyNames.get(i));}RLock lock this.redissonClient.getMultiLock(locks);boolean success lock.tryLock(timeout, TimeUnit.MILLISECONDS);// 加锁成功走业务逻辑if (success) {this.doBusiness(o);}// 释放锁unlockAfterTransaction(lock);}// 业务逻辑abstract Object doBusiness(Object... o);private void unlockAfterTransaction(RLock lock) {//事物完成后释放锁TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {Overridepublic void afterCompletion(int status) {super.afterCompletion(status);lock.unlock();}});} }当我们使用1.8以上的JDK时针对模板模式做了很多优化。我们可以将实现方法作为函数式方法传入模版中 public class LockFunctionService {RedissonClient redissonClient;public void lock(ListString keyNames, Long timeout, ILockCallback lockCallback) throws InterruptedException {// 加锁RLock[] locks new RLock[keyNames.size()];for(int i 0; i keyNames.size(); i) {locks[i] this.redissonClient.getLock(keyNames.get(i));}RLock lock this.redissonClient.getMultiLock(locks);boolean success lock.tryLock(timeout, TimeUnit.MILLISECONDS);// 加锁成功走业务逻辑if (success) {lockCallback.callback();}// 释放锁unlockAfterTransaction(lock);}private void unlockAfterTransaction(RLock lock) {//事物完成后释放锁TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {Overridepublic void afterCompletion(int status) {super.afterCompletion(status);lock.unlock();}});}}interface ILockCallbackT {T callback(); }当我们调用时比之前方便多了 public static void main(String[] args) throws InterruptedException {LockFunctionService lockFunctionService new LockFunctionService();ListString keys Lists.newArrayList(lock_1);// 需要加锁执行释放lockFunctionService.lock(keys, 5L, () - {System.out.println(执行业务逻辑);return null;}); }