云南省建设厅网站查询网页怎么制作

前言

  在此说明，本文章不只是讲一些抽象的概念，而是可落地的，在日常工作中基本上进行修改一下便可以使用。书接上回，上篇自研分布式锁的文章使用是一个自己手写的一个分布式锁，按照JUC里面java.util.concurrent.locks.Lock接口规范编写的。其主要逻辑为：
 lock()加锁关键逻辑
  1.加锁：加锁实际上就是在redis中，给key设置一个值，为了避免死锁，并给定一个过期时间；
  2.可重入：加锁的LUA脚本，通过redis里面的hash数据模型，加锁和可重入性都要保证；
  3.自旋：加锁不成，需要while进行重试并自旋，AQS；
  4.续期：在过期时间内，一定时间内业务还未完成，自动给锁续期；
  unlock()解锁关键逻辑
  将redis的key删除，但是也不能乱删，不能说客户端1的请求将客户端2的锁给删除掉了，只能自己删自己的锁；
  考虑可重入性的递减，加锁几次就需要删除几次，最后到零了，直接del删除；上面自研的redis锁对于一般中小公司，不是特别高并发场景足够用了，单机redis小业务也撑得住。

Redis分布式锁-Redlock红锁算法

  英文名称为： Distributed locks with Redis，官方说明文档:点击此处

为什么基于故障转移的实施是不够的

  上篇文章中自己手动写的分布式锁存在的问题是单点故障时会出现数据不安全。看看官方文档怎么说？
为了理解我们想要改进的地方，让我们分析一下基于多数Redis额分布式锁库的现状。使用Redis锁定资源的最简单方法是在实例中创建一个键。使用Redis的过期功能，秘钥通常是在有限的生存时间内创建的，因此最终它会被释放。当客户需要释放资源时，它会删除秘钥。
有个问题：如果Master宕机了怎么办？主从架构条件下，添加一个副本，这种方式是不可行的。这样做我们无法实现显示互斥的安全属性，因为Redis复制是异步的。
官方文档中的一段说明进行解释，此模型存在竞争条件：
  1.Client A获取master中的锁；
  2.在对秘钥的写入传输到副本之前，主服务器崩了；
  3.副本被提升为主节点；
  4.客户端B获取对同一资源A持有锁的锁；（违反安全规定）
  有时在特殊情况下，例如在故障期间，多个客户端可以同时持有锁是完全没有问题的。如果是这种情况，您可以使用基于复制的方案。否则，我们建议实施本文档中描述 解决方案。

  简单的描述就是，当线程1首先获取锁成功，将键值写入Redis的master节点中，在Redis将该键值对同步到slave节点之前，master发生了故障；Redis触发故障转移，其中一个slave升级为新的master，此时master并不包含线程1写入的键值对，因此线程2尝试获取锁也是可以成功拿到锁的，此时相当于有两个线程获取了锁，可能会导致各种预期之外的情况发生，例如最常见的脏数据。

  我们加的是排它独占锁，同一时间只能有一个建Redis锁成功并持有锁，严禁出现2个以上的请求线程拿到锁。

RedLock算法设计理念

  Redis之父提出了RedLock算法解决上面这个一锁被多建的问题
  Redis也提供了Redlock算法，用来实现基于多个实例的分布式锁。 锁变量由多个实例维护，即使有实例发生了故障，锁变量仍然是存在的，客户端还是可以完成锁操作。Redlock算法是实现高可靠分布式锁的一种有效解决方案，可以在实际开发中使用。

设计理念

  该方案也是基于(set加锁、Lua脚本解锁）进行改良的，所以redis之父antirez只描述了差异的地方，大致方案如下：假设我们有N个Redis主节点，例如N= 5这些节点是完全独立的，我们不使用复制或任何其他隐式协调系统，为了取到锁客户端执行以下操作：
  1.获取当前时间，以毫秒为单位;
  2.依次尝试从5个实例，使用相同的 key和随机值（例如 UUID）获取锁。当向Redis 请求获取锁时，客户端应该设置一个超时时间，这个超时时间应该小于锁的失效时间。例如你的锁自动失效时间为10秒，则超时时间应该在5-50毫秒之间。这样可以防止客户端在试图与一个宕机的 Redis 节点对话时长时间处于阻塞状态。如果一个实例不可用，客户端应该尽快尝试去另外一个Redis实例请求获取锁;
  3.客户端通过当前时间减去步骤1记录的时间来计算获取锁使用的时间。当且仅当从大多数(N/2+1，这里是3个节点)的Redis节点都取到锁，并且获取锁使用的时间小于锁失效时间时，锁才算获取成功;
  4.如果取到了锁，其真正有效时间等于初始有效时间减去获取锁所使用的时间（步骤3计算的结果）。
  5.如果由于某些原因未能获得锁（无法在至少N/2＋1个Redis实例获取锁、或获取锁的时间超过了有效时间)，客户端应该在所有的Redis 实例上进行解锁（即便某些Redis实例根本就没有加锁成功，防止某些节点获取到锁但是客户端没有得到响应而导致接下来的一段时间不能被重新获取锁）。该方案为了解决数据不一致的问题，直接舍弃了异步复制只使用master节点，同时由于舍弃了slave，为了保证可用性，引入了N个节点。客户端只有在满足下面的这两个条件时，才能认为是加锁成功。
  条件1:客户端从超过半数（大于等于N/2+1）的Redis实例上成功获取到了锁;
  条件2:客户端获取锁的总耗时没有超过锁的有效时间。
该方案为了解决数据不一致的问题，直接舍弃了异步复制只使用master节点，同时由于舍弃了从节点（slave），为了保证可用性，引入了N个节点，官方建议是5。

redi只支持 AP，即高可用，为了解决CP的风险，采用N个节点，N为奇数，上面的3个master个独立，不是主从复制。为什么是奇数？N=2X+1,其中N是最终部署主机数，X是容错主机数。

什么是容错

  失败了多少个机器实例后我还是可以容忍的，所谓容就是数据的一致性还是可以的，CP数据一致性还是可以满足，加入在集群环境中，redis失败1台，可以接受。2X + 1 = 2* 1 +1 = 3， 部署3台，死了1个剩下2个可以正常工作，那就部署3台。加入在集群环境中，redis失败1台，可以接受。2X + 1 = 2* 2 +1 = 5， 部署5台，死了2个剩下3个可以正常工作，那就部署5台。

为什么是奇数

  最少的机器，最多的效果。加入集群环境中，redis失败1台，可接受。2N + 2 = 2 * 1 + 4，部署4台。

使用Redisson进行编码改造

  可重入锁，基于Redis实现的Redisson分布式可重入锁RLock java实现了java.util.concurrent.lock.Lock接口。同时还提供异步、反射和RxJava2标准额接口。
  Spring集成Redission环境开发。bom文件中引入如下代码：

<dependency><groupId>org.redisson</groupId><artifactId>redisson</artifactId><version>3.13.4</version>
</dependency>

  配置类：RedisConfig

@Bean
public Redisson redisson() {Config config = new Config();config.useSingleServer().setAddress("redis://127.0.0.1:6379").setDatabase(0).setPassword("123456");return (Redisson) Redisson.create(config);
}

  修改服务方法：InventoryService（在日常工作中的业务成层类中）

@Autowired
private Redisson redisson;
public String saleByRedisson() {String resMessgae = "";RLock redissonLock = redisson.getLock("luojiaRedisLock");redissonLock.lock();try {// 1 抢锁成功，查询库存信息String result = stringRedisTemplate.opsForValue().get("inventory01");// 2 判断库存书否足够Integer inventoryNum = result == null ? 0 : Integer.parseInt(result);// 3 扣减库存，每次减少一个库存if (inventoryNum > 0) {stringRedisTemplate.opsForValue().set("inventory01", String.valueOf(--inventoryNum));resMessgae = "成功卖出一个商品，库存剩余：" + inventoryNum + "\t" + "，服务端口号：" + port;log.info(resMessgae);} else {resMessgae = "商品已售罄。" + "\t" + "，服务端口号：" + port;log.info(resMessgae);}} finally {redissonLock.unlock();}return resMessgae;
}

  仔细看上面代码依然存在释放锁的问题，可能存在删除其他线程的锁，所以在finally中添加如下方法。

finally {// 改进点，只能删除属于自己的key，不能删除别人的if(redissonLock.isLocked() && redissonLock.isHeldByCurrentThread()) {redissonLock.unlock();}
}

watch dog(看门狗)自动延期机制

  源码中初始化了一个定时器，dely的时间是 internalLockLeaseTime / 3。在Redisson中，internalLockLeaseTime 是获取配置的看门狗的时间，默认是30s，也就是每隔10s续期一次，每次重新设置过期时间为30s。总而言之，看门狗的本质在于开启一个监听线程，定期检查锁是否持有，有则延长过期时间。
Redisson看门狗续期源码：

private void renewExpiration() {ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());if (ee == null) {return;}Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {@Overridepublic void run(Timeout timeout) throws Exception {ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName());if (ent == null) {return;}Long threadId = ent.getFirstThreadId();if (threadId == null) {return;}RFuture<Boolean> future = renewExpirationAsync(threadId);future.onComplete((res, e) -> {if (e != null) {log.error("Can't update lock " + getName() + " expiration", e);return;}if (res) {// reschedule itselfrenewExpiration();}});}}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);ee.setTimeout(task);
}

结语

  本文章到此结束，麻烦路过点赞后续还会补充内容。