高性能分布式缓存Redis-分布式锁与布隆过滤器

一、分布式锁

我们先来看一下本地锁

在并发编程中，我们通过锁，来避免由于竞争而造成的数据不一致问题。通常，我们以 synchronized 、Lock 来使用它（单机情况）

来看这段代码

@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 something
            log.info("我抢到茅台了!");
            return "ok"; 
        }else { 
            return "no"; 
        }
    }
}

现象：本地锁在多节点下失效(集群/分布式)
原因：本地锁它只能锁住本地JVM进程中的多个线程，对于多个JVM进程的不同线程间是锁不住的
解决：分布式锁（在分布式环境下提供锁服务，并且达到本地锁的效果）

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

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

基于Redis实现分布式锁

锁的实现主要基于 redis 的 SETNX 命令：

大致流程：

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

锁超时

仅仅使用这个流程，会出现什么问题吗？思考：如果程序在第2步执行出现异常退出或宕机，没有执行第3步释放锁，岂不就使得这个锁一直未释放，别的请求一直得不到这个锁，这就产生了死锁问题！这也就是分布式锁的锁超时特点。

如何解决？在抢到锁的时候，给这个锁加上一个过期时间，即使没有执行手动释放锁，也会在过期时间到了自动释放锁！

代码如下：

    @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 something
                    log.info("我抢到茅台了!");
                    return "ok";
                }else {
                    return "no";
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                //释放锁
                redisTemplate.delete(lockey);
            }
        }
        return "dont get lock";
    }

原子性操作

在以上代码中，setnx和expire的操作是分开执行的，假设此时，setnx执行完毕，程序异常退出或宕机了，那就无法设置好过期时间，这样也会导致锁一直得不到释放

究其原因，这两步是非原子性操作（解决：2.6以前可用使用lua脚本，2.6以后可用set命令），使用lua脚本把这两个命令变成一气呵成的操作，确保都同时执行成功。

@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>() {
            @Override
            public 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 something
                    log.info("我抢到茅台了!");
                    return "ok";
                }else {
                    return "no";
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                //释放锁
                redisTemplate.delete(lockey);
            }
        }
        return "dont get lock";
    }

错误解锁

思考以上代码：假设设置的超时时间小于业务执行时间，导致业务代码没有执行完，锁就被释放了，这时候其他请求就能进到这个锁，操作一些共享变量导致线程安全问题

解决方案：在setnx时，将value设置成一个唯一标识，解锁的时候，要进行value的匹配，匹配上了才能解锁

@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>() {
            @Override
            public 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 something
                    log.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>() {
                    @Override
                    public 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";
    }

锁续期

思考上面代码：在下图里面，只是解决了业务代码执行完，释放锁只能释放自己的锁，但是，当锁过期了，其他请求不也一样的进到了锁里面操作共享变量吗？究其原因是锁的过期时间小于业务代码执行时间，那解决办法自然就是把这个时间延长呗

那这个时间延长到底如何做？

给拿到锁的线程创建一个守护线程(看门狗)，守护线程定时/延迟（如每隔5s检查业务是否执行完毕）判断拿到锁的线程是否还继续持有锁，如果持有则为其续期

    //模拟一下守护线程为其续期
    ScheduledExecutorService executorService;//创建守护线程池
    ConcurrentSkipListSet<String> set = new ConcurrentSkipListSet<String>();//队列

    @PostConstruct
    public void init2(){
        executorService = Executors.newScheduledThreadPool(1);

        //编写续期的lua
        String 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() {
            @Override
            public void run() {
                Iterator<String> iterator = set.iterator();
                while (iterator.hasNext()) {
                    String rquestid = iterator.next();

                    redisTemplate.execute(new RedisCallback<Boolean>() {
                        @Override
                        public 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>() {
                    @Override
                    public 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的类型来做这个。基于这个思路我们看一下实现流程：

加锁：当一个客户端尝试获取锁时，首先检查锁是否已被其他客户端持有。如果没有，则设置锁并成为持有者；如果有，但持有者是自己，则增加计数器。
解锁：当一个客户端释放锁时，减少计数器。如果计数器变为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>

    @Bean
    public RedissonClient redissonClient() {
        Config config = new Config();
        config.useSingleServer().setAddress("redis://"+host+":"+port);
        return Redisson.create(config);
    }

    @Autowired
    RedissonClient 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 something
                log.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");
    //构造Redisson
    RedissonClient redisson = Redisson.create(config);

    RBloomFilter<String> bloomFilter = redisson.getBloomFilter("phoneList");
    //初始化布隆过滤器：预计元素为100000000L,偏差率为3%
    bloomFilter.tryInit(100000000L,0.03);
    //将号码10086插入到布隆过滤器中
    bloomFilter.add("10086");

    //判断下面号码是否在布隆过滤器中
    //输出false
    System.out.println(bloomFilter.contains("123456"));
    //输出true
    System.out.println(bloomFilter.contains("10086"));
  }
}

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