一、分布式锁

我们先来看一下本地锁

@Autowired
RedisTemplate<String,String> redisTemplate;String maotai = "maotai20210321001";//茅台商品编号 @PostConstruct
public void init(){ //此处模拟向缓存中存入商品库存操作 redisTemplate.opsForValue().set(maotai,"100"); 
}@GetMapping("/get/maotai2")
public String seckillMaotai2() {synchronized (this) {Integer count = Integer.parseInt(redisTemplate.opsForValue().get(maotai)); // 1 //如果还有库存 if (count > 0) { //抢到了茅台，库存减一 redisTemplate.opsForValue().set(maotai,String.valueOf(count-1));//后续操作 do somethinglog.info("我抢到茅台了!");return "ok"; }else { return "no"; }}
}

那么，到底什么是分布式锁呢？

• 当在分布式架构下，数据只有一份（或有限制），此时需要利用锁的技术控制某一时刻修改数据的进程数。
• 用一个状态值表示锁，对锁的占用和释放通过状态值来标识。

基于Redis实现分布式锁

大致流程：

• 1. 使用 SETNX 命令获取锁，若返回0（key已存在，锁已存在）则获取失败，反之获取成功
• 2. 执行业务逻辑
• 3. 释放锁，使用 DEL 命令将锁数据删除

    @GetMapping("/get/maotai3")public String seckillMaotai3() {//获取锁Boolean islock = redisTemplate.opsForValue().setIfAbsent(lockey, "1");if (islock) {//设置锁的过期时间redisTemplate.expire(lockey,5, TimeUnit.SECONDS);try {Integer count = Integer.parseInt(redisTemplate.opsForValue().get(maotai)); // 1//如果还有库存if (count > 0) {//抢到了茅台，库存减一redisTemplate.opsForValue().set(maotai,String.valueOf(count-1));//后续操作 do somethinglog.info("我抢到茅台了!");return "ok";}else {return "no";}} catch (Exception e) {e.printStackTrace();} finally {//释放锁redisTemplate.delete(lockey);}}return "dont get lock";}

@GetMapping("/get/maotai4")public String seckillMaotai4() {//获取锁String locklua ="" +"if redis.call('setnx',KEYS[1],ARGV[1]) == 1 then redis.call('expire',KEYS[1],ARGV[2]) ; return true " +"else return false " +"end";Boolean islock = redisTemplate.execute(new RedisCallback<Boolean>() {@Overridepublic Boolean doInRedis(RedisConnection redisConnection) throws DataAccessException {Boolean eval = redisConnection.eval(locklua.getBytes(),ReturnType.BOOLEAN,1,lockey.getBytes(),requestid.getBytes(),"5".getBytes());return eval;}});if (islock) {try {Integer count = Integer.parseInt(redisTemplate.opsForValue().get(maotai)); // 1//如果还有库存if (count > 0) {//抢到了茅台，库存减一redisTemplate.opsForValue().set(maotai,String.valueOf(count-1));//后续操作 do somethinglog.info("我抢到茅台了!");return "ok";}else {return "no";}} catch (Exception e) {e.printStackTrace();} finally {//释放锁redisTemplate.delete(lockey);}}return "dont get lock";}

@GetMapping("/get/maotai4")public String seckillMaotai4() {String requestid = UUID.randomUUID().toString() + Thread.currentThread().getId();/*String locklua ="" +"if redis.call('setnx',KEYS[1],ARGV[1]) == 1 then redis.call('expire',KEYS[1],ARGV[2]) ; return true " +"else return false " +"end";Boolean islock = redisTemplate.execute(new RedisCallback<Boolean>() {@Overridepublic Boolean doInRedis(RedisConnection redisConnection) throws DataAccessException {Boolean eval = redisConnection.eval(locklua.getBytes(),ReturnType.BOOLEAN,1,lockey.getBytes(),requestid.getBytes(),"5".getBytes());return eval;}});*///获取锁Boolean islock = redisTemplate.opsForValue().setIfAbsent(lockey,requestid,5,TimeUnit.SECONDS);if (islock) {try {Integer count = Integer.parseInt(redisTemplate.opsForValue().get(maotai)); // 1//如果还有库存if (count > 0) {//抢到了茅台，库存减一redisTemplate.opsForValue().set(maotai,String.valueOf(count-1));//后续操作 do somethinglog.info("我抢到茅台了!");return "ok";}else {return "no";}} catch (Exception e) {e.printStackTrace();} finally {//释放锁//判断是自己的锁才能去释放 这种操作不是原子性的/*String id = redisTemplate.opsForValue().get(lockey);if (id !=null && id.equals(requestid)) {redisTemplate.delete(lockey);}*/String unlocklua = "" +"if redis.call('get',KEYS[1]) == ARGV[1] then redis.call('del',KEYS[1]) ; return true " +"else return false " +"end";redisTemplate.execute(new RedisCallback<Boolean>() {@Overridepublic Boolean doInRedis(RedisConnection redisConnection) throws DataAccessException {Boolean eval = redisConnection.eval(unlocklua.getBytes(),ReturnType.BOOLEAN,1,lockey.getBytes(),requestid.getBytes());return eval;}});}}return "dont get lock";}

    //模拟一下守护线程为其续期ScheduledExecutorService executorService;//创建守护线程池ConcurrentSkipListSet<String> set = new ConcurrentSkipListSet<String>();//队列@PostConstructpublic void init2(){executorService = Executors.newScheduledThreadPool(1);//编写续期的luaString expirrenew = "" +"if redis.call('get',KEYS[1]) == ARGV[1] then redis.call('expire',KEYS[1],ARGV[2]) ; return true " +"else return false " +"end";executorService.scheduleAtFixedRate(new Runnable() {@Overridepublic void run() {Iterator<String> iterator = set.iterator();while (iterator.hasNext()) {String rquestid = iterator.next();redisTemplate.execute(new RedisCallback<Boolean>() {@Overridepublic Boolean doInRedis(RedisConnection redisConnection) throws DataAccessException {Boolean eval = false;try {eval = redisConnection.eval(expirrenew.getBytes(),ReturnType.BOOLEAN,1,lockey.getBytes(),rquestid.getBytes(),"5".getBytes());} catch (Exception e) {log.error("锁续期失败,{}",e.getMessage());}return eval;}});}}},0,1,TimeUnit.SECONDS);}@GetMapping("/get/maotai5")public String seckillMaotai5() {String requestid = UUID.randomUUID().toString() + Thread.currentThread().getId();//获取锁Boolean islock = redisTemplate.opsForValue().setIfAbsent(lockey,requestid,5,TimeUnit.SECONDS);if (islock) {//获取锁成功后让守护线程为其续期set.add(requestid);try {Integer count = Integer.parseInt(redisTemplate.opsForValue().get(maotai)); // 1//如果还有库存if (count > 0) {//抢到了茅台，库存减一redisTemplate.opsForValue().set(maotai,String.valueOf(count-1));//后续操作 do something//seckillMaotai5();//模拟业务超时TimeUnit.SECONDS.sleep(10);log.info("我抢到茅台了!");return "ok";}else {return "no";}} catch (Exception e) {e.printStackTrace();} finally {//解除锁续期set.remove(requestid);//释放锁String unlocklua = "" +"if redis.call('get',KEYS[1]) == ARGV[1] then redis.call('del',KEYS[1]) ; return true " +"else return false " +"end";redisTemplate.execute(new RedisCallback<Boolean>() {@Overridepublic Boolean doInRedis(RedisConnection redisConnection) throws DataAccessException {Boolean eval = redisConnection.eval(unlocklua.getBytes(),ReturnType.BOOLEAN,1,lockey.getBytes(),requestid.getBytes());return eval;}});}}return "dont get lock";}

锁的可重入

思考一个问题，如果在一段代码里，我拿到了锁，这个代码要递归调用自己，就面临着再次获取锁，所以确保这次获取锁是成功的，这称为锁的可重入

可重入如何做呢？我们可不可以使得value是一个数值，在锁的内部每加一次锁就让数值加1，每释放一次锁就让数值减1。我们用hash的类型来做这个。基于这个思路我们看一下实现流程：

1. 加锁：当一个客户端尝试获取锁时，首先检查锁是否已被其他客户端持有。如果没有，则设置锁并成为持有者；如果有，但持有者是自己，则增加计数器。
2. 解锁：当一个客户端释放锁时，减少计数器。如果计数器变为0，则删除锁，这样其他客户端才能获取锁。

 加锁的lua脚本：

                        "if (redis.call('exists', KEYS[1]) == 0) then " +#设置锁key,field是唯一标识,value是重入次数"redis.call('hset', KEYS[1], ARGV[2], 1); " +#设置锁key的过期时间 默认30s"redis.call('pexpire', KEYS[1], ARGV[1]); " +"return nil; " +"end; " +#如果锁key存在"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +#重入次数+1"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +#重置过期时间"redis.call('pexpire', KEYS[1], ARGV[1]); " +"return nil; " +"end; " +"return redis.call('pttl', KEYS[1]);"

解锁的lua脚本：

                      "if (redis.call('exists', KEYS[1]) == 0) then " +"redis.call('publish', KEYS[2], ARGV[1]); " +"return 1; " +"end;" +// hash 中的field 不存在时直接返回,"if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +"return nil;" +"end; " +"local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +//重入次数-1后如果还大于0,延长过期时间"if (counter > 0) then " +"redis.call('pexpire', KEYS[1], ARGV[2]); " +"return 0; " +"else " +//重入次数-1后如果归0,则删除key，并向redisson_lock__channel:{key}频道发布锁释放消息"redis.call('del', KEYS[1]); " +"redis.call('publish', KEYS[2], ARGV[1]); " +"return 1; "+"end; " +"return nil;"

阻塞锁

我们要使得，每个请求来获取锁，如果锁已被占用，那么获取不到就等待锁的释放，直到获取到锁或者等待超时，常见方案有两种：

• 1：基于客户端轮询的方案
• 2：基于redis的发布/订阅方案

第2种方案：对于每个抢不到锁的进程，就订阅一个频道，当释放锁时，会向这个频道发布通知， 收到通知再进行重新抢锁

以上这些特性都是分布式锁应该满足的，那么自己写起来还是不太方便，Redisson就帮我们封装好了这一切，直接用

基于Redisson 实现分布式锁

Redisson的介绍

        Redisson内置了一系列的分布式对象，分布式集合，分布式锁，分布式服务等诸多功能特性，是一款基于Redis实现，拥有一系列分布式系统功能特性的工具包，是实现分布式系统架构中缓存中间件的最佳选择。

使用步骤

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

    @Beanpublic RedissonClient redissonClient() {Config config = new Config();config.useSingleServer().setAddress("redis://"+host+":"+port);return Redisson.create(config);}@AutowiredRedissonClient redissonClient;@GetMapping("/get/maotai6")public String seckillMaotai6() {//要去获取锁RLock lock = redissonClient.getLock(lockey);lock.lock();try {Integer count = Integer.parseInt(redisTemplate.opsForValue().get(maotai)); // 1//如果还有库存if (count > 0) {//抢到了茅台，库存减一redisTemplate.opsForValue().set(maotai,String.valueOf(count-1));//后续操作 do somethinglog.info("我抢到茅台了!");return "ok";}else {return "no";}} catch (Exception e) {e.printStackTrace();} finally {lock.unlock();;}return "";}

分布式锁特点

总结来说，设计一个完善的分布式锁，需要满足下面这些特点

• 互斥性：不仅要在同一jvm进程下的不同线程间互斥，更要在不同jvm进程下的不同线程间互斥
• 锁超时：支持锁的自动释放，防止死锁
• 正确，高效，高可用（解决错误解锁问题）：解铃还须系铃人（加锁和解锁必须是同一个线程），加锁和解锁操作一定要高效，提供锁的服务要具备容错性
• 可重入：如果一个线程拿到了锁之后继续去获取锁还能获取到，我们称锁是可重入的（方法的递归调用）
• 阻塞/非阻塞：如果获取不到直接返回视为非阻塞的，如果获取不到会等待锁的释放直到获取锁或者等待超时，视为阻塞的
• 公平/非公平：按照请求的顺序获取锁视为公平的

前三点使我们必须要满足的，后两点是分布式锁的类型 

二、布隆过滤器

我们先从一个场景说起

如果数据在缓存和数据库中都不存在，以id=-1为例，如果频繁的请求这个id的数据，会被频繁的访问数据库判断是否存在，直至请求多到把数据库压垮，这就是缓存穿透问题

那如何解决呢？我们就可以用布隆过滤器，布隆过滤器就可以实现在海量元素中，快速判断一个元素是否存在

布隆过滤器本质上其实就是一个很长的二进制向量和一系列随机映射函数。专门用来检测集合中是否存在特定的元素

布隆过滤器的设计

BF是由一个长度为m比特的位数组（bit array）与k个哈希函数（hash function）组成的数据结构。位数组均初始化为0，所有哈希函数都可以分别把输入数据尽量均匀地散列。

这个二进制位数据就由bitmap来实现

如果我们要映射一个值到布隆过滤器中，我们需要使用多个不同的哈希函数生成多个哈希值，并对每个生成的哈希值指向的 bit 位，设置为1

 当要插入一个元素时，将其数据分别输入k个哈希函数，产生k个哈希值。以哈希值作为位数组中的下标，将所有k个对应的比特置为1。

当要查询（即判断是否存在）一个元素时，同样将其数据输入哈希函数，然后检查对应的k个比特。如果有任意一个比特为0，表明该元素一定不在集合中。如果所有比特均为1，表明该集合有（较大的）可能性在集合中。为什么不是一定在集合中呢？因为一个比特被置为1有可能会受到其他元素的影响，这就是所谓“假阳性”（false positive）。相对地，“假阴性”（false negative）在BF中是绝不会出现的。

总结，对于BF的查询结果：

• 如果这些点有任何一个 0，则被检索元素一定不在；
• 如果都是 1，则被检索元素很可能在。

 布隆过滤器的误判是指多个输入经过哈希之后在相同的bit位置1了，这样就无法判断究竟是哪个输入产生的，因此误判的根源在于相同的 bit 位被多次映射且置 1。

 误报率指的是你愿意接受的误报概率。误报是指布隆过滤器告诉你一个元素“可能”存在于集合中，但实际上并不在集合中。误报率越低，布隆过滤器所需的内存就越多。

在redis中使用BF

<dependency><groupId>org.redisson</groupId><artifactId>redisson-spring-boot-starter</artifactId><version>3.13.4</version>
</dependency>

public class RedissonBloomFilter {public static void main(String[] args) {Config config = new Config();config.useSingleServer().setAddress("redis://127.0.0.1:6379");config.useSingleServer().setPassword("1234");//构造RedissonRedissonClient redisson = Redisson.create(config);RBloomFilter<String> bloomFilter = redisson.getBloomFilter("phoneList");//初始化布隆过滤器：预计元素为100000000L,偏差率为3%bloomFilter.tryInit(100000000L,0.03);//将号码10086插入到布隆过滤器中bloomFilter.add("10086");//判断下面号码是否在布隆过滤器中//输出falseSystem.out.println(bloomFilter.contains("123456"));//输出trueSystem.out.println(bloomFilter.contains("10086"));}
}

解释一下bloomFilter.tryInit(100000000L,0.03);

• 100000000L 表示你期望这个布隆过滤器能够容纳大约一亿（100,000,000）个元素。
• 0.03 表示你希望这个布隆过滤器误判率不超过 3%。