1、布隆过滤

1.1、简介

        由一个初值都为零的bit数组和多个哈希函数构成，可以用来快速判断集合中是否存在某个元素，减少占用内存，不保存数据信息，只是在内存中做出一个标记。

        它实际上是一个很长的二进制数组(00000000)+一系列随机hash算法映射函数，主要用于判断一个元素是否在集合中。通常我们会遇到很多要判断一个元素是否在某个集合中的业务场景，一般想到的是将集合中所有元素保存起来，然后通过比较确定。

        使用布隆过滤的话一个元素如果判断结果：存在时，元素不一定存在，但是判断结果为不存在时，则一定不存在。布隆过滤器可以添加元素，但是不能删除元素，由于涉及hashcode判断依据，删掉元素会导致误判率增加。

        布隆过滤器(Bloom Filter) 是一种专门用来解决去重问题的高级数据结构。实质就是一个大型*位数组*和几个不同的无偏hash函数(无偏表示分布均匀)。由一个初值都为零的bit数组和多个个哈希函数构成，用来快速判断某个数据是否存在。但是跟 HyperLogLog 一样，它也一样有那么一点点不精确，也存在一定的误判概率

1.2、使用特点

        添加key时使用多个hash函数对key进行hash运算得到一个整数索引值，对位数组长度进行取模运算得到一个位置，每个hash函数都会得到一个不同的位置，将这几个位置都置1就完成了add操作。

        当有变量被加入集合时，通过N个映射函数将这个变量映射成位图中的N个点，把它们置为 1（假定有两个变量都通过 3 个映射函数）。

        查询key时只要有其中一位是零就表示这个key不存在，但如果都是1，则不一定存在对应的key。如果这些点，有任何一个为零则被查询变量一定不在，如果都是 1，则被查询变量很可能存在，为什么说是可能存在，而不是一定存在呢？那是因为映射函数本身就是散列函数，散列函数是会有碰撞的。（见上图3号坑两个对象都1）

        此外还有可能出现的误差是因为，哈希函数的概念是：将任意大小的输入数据转换成特定大小的输出数据的函数，转换后的数据称为哈希值或哈希编码，也叫散列值。

        如果两个散列值是不相同的（根据同一函数）那么这两个散列值的原始输入也是不相同的。

这个特性是散列函数具有确定性的结果，具有这种性质的散列函数称为单向散列函数。

        散列函数的输入和输出不是唯一对应关系的，如果两个散列值相同，两个输入值很可能是相同的，但也可能不同，这种情况称为“散列碰撞（collision）”。用 hash表存储大数据量时，空间效率还是很低，当只有一个 hash 函数时，还很容易发生哈希碰撞。

1.3、使用步骤

1.3.1、初始化bitmap

        布隆过滤器 本质上 是由长度为 m 的位向量或位列表（仅包含 0 或 1 位值的列表）组成，最初所有的值均设置为 0。

1.3.2、添加占坑位

当我们向布隆过滤器中添加数据时，为了尽量地址不冲突，会使用多个 hash 函数对 key 进行运算，算得一个下标索引值，然后对位数组长度进行取模运算得到一个位置，每个 hash 函数都会算得一个不同的位置。再把位数组的这几个位置都置为 1 就完成了 add 操作。

例如，我们添加一个字符串wmyskxz，对字符串进行多次hash(key) → 取模运行→ 得到坑位

1.3.3、判断是否存在

向布隆过滤器查询某个key是否存在时，先把这个 key 通过相同的多个 hash 函数进行运算，查看对应的位置是否都为 1，只要有一个位为零，那么说明布隆过滤器中这个 key 不存在；如果这几个位置全都是 1，那么说明极有可能存在；因为这些位置的 1 可能是因为其他的 key 存在导致的，也就是前面说过的hash冲突。。。。。

就比如我们在 add 了字符串wmyskxz数据之后，很明显下面1/3/5 这几个位置的 1 是因为第一次添加的 wmyskxz 而导致的；此时我们查询一个没添加过的不存在的字符串inexistent-key，它有可能计算后坑位也是1/3/5 ，这就是误判了

2、缓存问题

2.1、缓存预热

        Redis缓存预热是指在服务器启动或应用程序启动之前，将一些数据先存储到Redis中，以提高Redis的性能和数据一致性。这可以减少服务器在启动或应用程序启动时的数据传输量和延迟，从而提高应用程序的性能和可靠性。 （1）数据准备

        在应用程序启动或服务器启动之前，准备一些数据，这些数据可以是静态数据、缓存数据或其他需要预热的数据。

（2）数据存储

        将数据存储到Redis中，可以使用Redis的列表（List）数据类型或集合（Set）数据类型。

（3）数据预热

        在服务器启动或应用程序启动之前，将数据存储到Redis中。可以使用Redis的客户端工具或命令行工具来执行此操作。

（4）数据清洗

        在服务器启动或应用程序启动之后，可能会对存储在Redis中的数据进行清洗和处理。例如，可以删除过期的数据、修改错误的数据等。

        需要注意的是，Redis缓存预热可能会增加服务器的开销，因此应该在必要时进行。同时，为了减少预热的次数，可以考虑使用Redis的其他数据类型，如哈希表（Hash）或有序集合（Sorted Set）。此外，为了提高数据一致性和性能，可以使用Redis的持久化功能，将数据存储到Redis中，并在服务器重启后自动恢复数据。

2.2、缓存雪崩

        缓存雪崩是指在同一时间大量的缓存key同时失效或者Redis的服务器宕机，然后大量的请求到达数据库，然后带来很大的压力。

2.2.1、解决方法

2.2.1.1、 设置有效期均匀分布

        将缓存的有效期设置为均匀分布，避免大量缓存同时失效。大概的意思就是在缓存数据的时候不要将缓存的数据存在Redis中的设置的过期是时间统一，应该让这些缓存的数据的过期时间分散开来，这样就不会导致在同一时间有大量的key过期。

2.2.1.2、 数据预热

        在系统上线前，将可能会访问的数据预先加载到缓存中，避免缓存冷启动，这个做法还是有必要的，因为这样可以避免大量访问请求没有命中Redis后去访问数据库。

2.2.1.3、 保证Redis服务高可用

        保证Redis服务的高可用性，避免Redis宕机的情况，这个方式也很好理解，Redis可以配置哨兵，用来监控Master主机的是否可以正常工作，如果宕机选出新的主机，然后如果是想要提高读写性能的话也可以配置成集群的模式，多台主机可以执行写操作，即使某一台主机挂掉然后还有其他的主机可以执行。

2.2.1.4、多级缓存

2.2.1.5、服务降级

2.3、缓存穿透

        首先说一下缓存穿透，简单的讲就是请求的数据首先在Redis中查询没有找到，然后请求数据库去访问数据，但是在数据库中也没有找到，但是用户知道自己查询的数据不存在，所以此后的所有的请求的压力都会到数据库中， 但是事实上请求也没有预期的结果。

2.3.1、解决方法

2.3.1.1、 业务层校验

        在缓存中没有命中的情况下，可以在业务层进行校验，如果校验不通过，直接返回错误信息，避免继续访问数据库。

        还有一种做法就是，没有命中Redis和数据库这时候后就可以使用null来应对这种情况，大概的思路就是当第一次访问数据库时如果没有数据的话，然后将null存入Redis， 这样就可以解决下一次数据命中的是Redis而不是数据库，然后判断Redis命中如果是空值的话就要返回错误信息，但是对于这种的空值的设置的话，可以设置超时时间自动清除，不建议设置的时间过长， 这样会导致数据的不一致的时间过长，如果数据库中的数据更新但是Redis中的数据还是null，导致请求的参数无法获取正确值，这种方式的优点就是简单方便操作，但是这样会额外消耗内存以及数据不一致(所以要设置过期时间短一点) 。

        还有对于一些访问数据的格式做出校验，不要让别人直接可以猜到数据然后直接大量的请求进行访问。此外是对于用户的权限验证。没有权限就直接拦截不让他访问。

2.3.1.2、 布隆过滤器

        布隆过滤器是一种数据结构，可以用于判断一个元素是否在一个集合中。在缓存中没有命中的情况下，可以使用布隆过滤器判断该数据是否存在，如果不存在，直接返回错误信息。

#初始化
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
​
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
​
/*** @auther zzyy* @create 2022-12-27 14:55* 布隆过滤器白名单初始化工具类，一开始就设置一部分数据为白名单所有，* 白名单业务默认规定：布隆过滤器有，redis也有。*/
@Component
@Slf4j
public class BloomFilterInit
{@Resourceprivate RedisTemplate redisTemplate;
​@PostConstruct//初始化白名单数据，故意差异化数据演示效果......public void init(){//白名单客户预加载到布隆过滤器String uid = "customer:12";//1 计算hashcode，由于可能有负数，直接取绝对值int hashValue = Math.abs(uid.hashCode());//2 通过hashValue和2的32次方取余后，获得对应的下标坑位long index = (long) (hashValue % Math.pow(2, 32));log.info(uid+" 对应------坑位index:{}",index);//3 设置redis里面bitmap对应坑位，该有值设置为1redisTemplate.opsForValue().setBit("whitelistCustomer",index,true);}
}
​
#utils
​
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
​
import javax.annotation.Resource;
​
/*** @auther zzyy* @create 2022-12-27 14:56*/
@Component
@Slf4j
public class CheckUtils
{@Resourceprivate RedisTemplate redisTemplate;
​public boolean checkWithBloomFilter(String checkItem,String key){int hashValue = Math.abs(key.hashCode());long index = (long) (hashValue % Math.pow(2, 32));boolean existOK = redisTemplate.opsForValue().getBit(checkItem, index);log.info("----->key:"+key+"\t对应坑位index:"+index+"\t是否存在:"+existOK);return existOK;}
}
​
​
#controller
​
import com.atguigu.redis7.entities.Customer;
import com.atguigu.redis7.service.CustomerSerivce;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;
​
import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.Random;
import java.util.Date;
import java.util.concurrent.ExecutionException;
​
/*** @auther zzyy* @create 2022-07-23 13:55*/
@Api(tags = "客户Customer接口+布隆过滤器讲解")
@RestController
@Slf4j
public class CustomerController
{@Resource private CustomerSerivce customerSerivce;
​@ApiOperation("数据库初始化2条Customer数据")@RequestMapping(value = "/customer/add", method = RequestMethod.POST)public void addCustomer() {for (int i = 0; i < 2; i++) {Customer customer = new Customer();
​customer.setCname("customer"+i);customer.setAge(new Random().nextInt(30)+1);customer.setPhone("1381111xxxx");customer.setSex((byte) new Random().nextInt(2));customer.setBirth(Date.from(LocalDateTime.now().atZone(ZoneId.systemDefault()).toInstant()));
​customerSerivce.addCustomer(customer);}}
​@ApiOperation("单个用户查询，按customerid查用户信息")@RequestMapping(value = "/customer/{id}", method = RequestMethod.GET)public Customer findCustomerById(@PathVariable int id) {return customerSerivce.findCustomerById(id);}
​@ApiOperation("BloomFilter案例讲解")@RequestMapping(value = "/customerbloomfilter/{id}", method = RequestMethod.GET)public Customer findCustomerByIdWithBloomFilter(@PathVariable int id) throws ExecutionException, InterruptedException{return customerSerivce.findCustomerByIdWithBloomFilter(id);}
}
​
​
​
​
#service
​
import com.atguigu.redis7.entities.Customer;
import com.atguigu.redis7.mapper.CustomerMapper;
import com.atguigu.redis7.utils.CheckUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
​
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
​
/*** @auther zzyy* @create 2022-07-23 13:55*/
@Service
@Slf4j
public class CustomerSerivce
{public static final String CACHE_KEY_CUSTOMER = "customer:";
​@Resourceprivate CustomerMapper customerMapper;@Resourceprivate RedisTemplate redisTemplate;
​@Resourceprivate CheckUtils checkUtils;
​public void addCustomer(Customer customer){int i = customerMapper.insertSelective(customer);
​if(i > 0){//到数据库里面，重新捞出新数据出来，做缓存customer=customerMapper.selectByPrimaryKey(customer.getId());//缓存keyString key=CACHE_KEY_CUSTOMER+customer.getId();//往mysql里面插入成功随后再从mysql查询出来，再插入redisredisTemplate.opsForValue().set(key,customer);}}
​public Customer findCustomerById(Integer customerId){Customer customer = null;
​//缓存key的名称String key=CACHE_KEY_CUSTOMER+customerId;
​//1 查询rediscustomer = (Customer) redisTemplate.opsForValue().get(key);
​//redis无，进一步查询mysqlif(customer==null){//2 从mysql查出来customercustomer=customerMapper.selectByPrimaryKey(customerId);// mysql有，redis无if (customer != null) {//3 把mysql捞到的数据写入redis，方便下次查询能redis命中。redisTemplate.opsForValue().set(key,customer);}}return customer;}
​/*** BloomFilter → redis → mysql* 白名单：whitelistCustomer* @param customerId* @return*/
​@Resourceprivate CheckUtils checkUtils;public Customer findCustomerByIdWithBloomFilter (Integer customerId){Customer customer = null;
​//缓存key的名称String key = CACHE_KEY_CUSTOMER + customerId;
​//布隆过滤器check，无是绝对无，有是可能有//===============================================if(!checkUtils.checkWithBloomFilter("whitelistCustomer",key)){log.info("白名单无此顾客信息:{}",key);return null;}//===============================================
​//1 查询rediscustomer = (Customer) redisTemplate.opsForValue().get(key);//redis无，进一步查询mysqlif (customer == null) {//2 从mysql查出来customercustomer = customerMapper.selectByPrimaryKey(customerId);// mysql有，redis无if (customer != null) {//3 把mysql捞到的数据写入redis，方便下次查询能redis命中。redisTemplate.opsForValue().set(key, customer);}}return customer;}
}

        比如设置白名单，或者每日签到，再者是打卡等等，在数据存储的时候可以将该数据的key使用hash函数获取索引，然后结合bitmap将该索引的位设置成为true（1），然后在数据读取的时候，首先在布隆过滤器中拦截查询的key，并且计算出这个key的索引值，然后读取redis中的bitmap结构中该索引位置的数据是true还是false，如果为true说明可能存在这个key所对应的数据，但是如果为false则说明这个key一定不存在，直接拦截！

2.3.1.3、Guava布隆过滤器

​

import com.atguigu.redis7.service.GuavaBloomFilterService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;
​
import javax.annotation.Resource;
​
/*** @auther zzyy* @create 2022-12-30 16:50*/
@Api(tags = "google工具Guava处理布隆过滤器")
@RestController
@Slf4j
public class GuavaBloomFilterController
{@Resourceprivate GuavaBloomFilterService guavaBloomFilterService;
​@ApiOperation("guava布隆过滤器插入100万样本数据并额外10W测试是否存在")@RequestMapping(value = "/guavafilter",method = RequestMethod.GET)public void guavaBloomFilter(){guavaBloomFilterService.guavaBloomFilter();}
}
​
​
​
import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnels;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
​
import java.util.ArrayList;
import java.util.List;
​
/*** @auther zzyy* @create 2022-12-30 16:50*/
@Service
@Slf4j
public class GuavaBloomFilterService{public static final int _1W = 10000;//布隆过滤器里预计要插入多少数据public static int size = 100 * _1W;//误判率,它越小误判的个数也就越少(思考，是不是可以设置的无限小，没有误判岂不更好)//fpp the desired false positive probabilitypublic static double fpp = 0.03;// 构建布隆过滤器private static BloomFilter<Integer> bloomFilter = BloomFilter.create(Funnels.integerFunnel(), size,fpp);public void guavaBloomFilter(){//1 先往布隆过滤器里面插入100万的样本数据for (int i = 1; i <=size; i++) {bloomFilter.put(i);}//故意取10万个不在过滤器里的值，看看有多少个会被认为在过滤器里List<Integer> list = new ArrayList<>(10 * _1W);for (int i = size+1; i <= size + (10 *_1W); i++) {if (bloomFilter.mightContain(i)) {log.info("被误判了:{}",i);list.add(i);}}log.info("误判的总数量：:{}",list.size());}
}

2.4、缓存击穿

        缓存击穿问题也叫做热点key问题，就是一个高并发访问并且缓存重建业务比较复杂的key突然失效啦，无数的请求访问会在瞬间给数据库带来巨大的冲击。

2.4.1、解决方法

2.4.1.1、设置热点数据永不过期

        将热点数据设置为永不过期，这样可以避免热点数据失效的情况，既然你是热点key而且我还害怕你过期，奶奶的给你设置为永不过期！！！

2.4.1.2、差异失效时间

        这种机制的话可以设置一个缓存的过期时间的不一致性，就是可以实现有时间的间隔来执行缓存更新数据，即使在某一时刻a缓存过期了还有b缓存用于命中旧的数据！

2.4.1.3、互斥锁

        在热点数据失效的情况下，可以使用互斥锁，保证只有一个线程去访问数据库，其他线程等待。相较于原来从缓存中查询不到数据后直接查询数据库而言，现在的方案是 进行查询之后，如果从缓存没有查询到数据，则进行互斥锁的获取，获取互斥锁后，判断是否获得到了锁，如果没有获得到，则休眠，过一会再进行尝试，直到获取到锁为止，才能进行查询

        如果获取到了锁的线程，再去进行查询，查询后将数据写入redis，再释放锁，返回数据，利用互斥锁就能保证只有一个线程去执行操作数据库的逻辑，防止缓存击穿

2.4.1.4、逻辑过期

        当用户开始查询redis时，判断是否命中，如果没有命中则直接返回空数据，不查询数据库，而一旦命中后，将value取出，判断value中的过期时间是否满足，如果没有过期，则直接返回redis中的数据，如果过期，则在开启独立线程后直接返回之前的数据，独立线程去重构数据，重构完成后释放互斥锁。

3、Redis锁

        在单机的系统中同一个JVM虚拟机内可以使用synchronized或者lock接口来实现锁的功能，但是对于分布式的功能来讲，里面有者不同的JVM虚拟机，这就到导致之前的在JVM内部的锁的无法使用，所以我们要能够有一种分布式的锁来完成这种功能。

那么要成为一个靠谱的分布式的锁需要具备那些的前提条件：

• 独占性 任何时刻有且只有一个线程所持有

• 高可用 在redis集群的环境下，不能出现某一个节点挂掉而出现获取锁和释放锁失败的情况，在高并发的情况下依然耐造

• 防死锁 杜绝死锁，必须有超时控制机制或者撤销机制，有一个兜底终止跳出的方案

• 不乱抢 自己的锁自己释放，不能二话不说释放别人的锁

• 重入性 同一个节点的同一个线程如果获取锁之后，它可以再次获取这个锁

3.1、使用单机的redis锁

public String sale(){String retMessage = "";lock.lock();try{//1 查询库存信息String result = stringRedisTemplate.opsForValue().get("inventory001");//2 判断库存是否足够Integer inventoryNumber = result == null ? 0 : Integer.parseInt(result);//3 扣减库存if(inventoryNumber > 0) {stringRedisTemplate.opsForValue().set("inventory001",String.valueOf(--inventoryNumber));retMessage = "成功卖出一个商品，库存剩余: "+inventoryNumber;System.out.println(retMessage);}else{retMessage = "商品卖完了，o(╥﹏╥)o";}}finally {lock.unlock();}return retMessage+"\t"+"服务端口号："+port;}

使用nginx配置负载均衡（暂时没有学习，先使用后补课）

        重新加载配置文件后启动服务器，然后使用jmeter进行高并发的测试。创建线程组，以及多少线程和持续时间，然后配置http请求，启动后看测式结果。我们再redsi中存储100个数据，然后两台服务器按道理说最后执行完毕刚好全部卖完，但是事实上redis中还剩余11个，另外在控制台中可以看出有卖出重复票的情况。

        那么使用单机锁来应对分布式高并发的项目来讲，就会出现锁不管用的情况。在单机环境下，可以使用synchronized或Lock来实现。

        但是在分布式系统中，因为竞争的线程可能不在同一个节点上（同一个jvm中），所以需要一个让所有进程都能访问到的锁来实现(比如redis或者zookeeper来构建)。不同进程jvm层面的锁就不管用了，那么可以利用第三方的一个组件，来获取锁，未获取到锁，则阻塞当前想要运行的线程。

3.2、分布式Redis锁

3.2.1、使用递归的方式实现

 /*** 分发布式锁 递归方式重试,容易出现栈溢出,高并发情况下建议使用while判断而不是if* @return*//*public String sale(){String retMessage="";String key="RedisLock";String uuidValue= IdUtil.simpleUUID()+":"+Thread.currentThread().getId();
​
​Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(key, uuidValue);
​if (!BooleanUtil.isFalse(flag)){//获取锁失败重试try {TimeUnit.MICROSECONDS.sleep(20);}catch (InterruptedException e){e.printStackTrace();}sale();}else {//获取锁成功try{//1 查询库存信息String result = stringRedisTemplate.opsForValue().get("inventory001");//2 判断库存是否足够Integer inventoryNumber = result == null ? 0 : Integer.parseInt(result);//3 扣减库存if(inventoryNumber > 0) {stringRedisTemplate.opsForValue().set("inventory001",String.valueOf(--inventoryNumber));retMessage = "成功卖出一个商品，库存剩余: "+inventoryNumber;System.out.println(retMessage);}else{retMessage = "商品卖完了，o(╥﹏╥)o";}}finally {//释放锁stringRedisTemplate.delete(key);}}
​return retMessage+"\t"+"服务端口号："+port;
​}*/

        这种方式其实效果不好而且也会出现超卖的情况，压力测试的时候数据只能执行到一半，然后中途停止，抢购的数据的话出现很多重复的数据。

3.2.2、自旋加while判断

 /*** 自旋，while* @return*/public String sale() {String retMessage = "";String key = "RedisLock";String uuidValue = IdUtil.simpleUUID() + ":" + Thread.currentThread().getId();
​
​//不适用递归而是使用自璇的方式,改为whilewhile (!stringRedisTemplate.opsForValue().setIfAbsent(key, uuidValue)) {//没有获取
​//获取锁失败重试try {TimeUnit.MICROSECONDS.sleep(20);}catch (InterruptedException e){e.printStackTrace();}}//抢锁成功!try{//1 查询库存信息String result = stringRedisTemplate.opsForValue().get("inventory001");//2 判断库存是否足够Integer inventoryNumber = result == null ? 0 : Integer.parseInt(result);//3 扣减库存if(inventoryNumber > 0) {stringRedisTemplate.opsForValue().set("inventory001",String.valueOf(--inventoryNumber));retMessage = "成功卖出一个商品，库存剩余: "+inventoryNumber;System.out.println(retMessage);}else{retMessage = "商品卖完了，o(╥﹏╥)o";}}finally {//释放锁stringRedisTemplate.delete(key);}return retMessage + "\t" + "服务端口号：" + port;
​}

        然后使用jemter压力测试，发现没有出现超卖的情况！但是随之而来的是当A线程执行完扣减商品的业务的时候突然宕机或者阻塞了很久，还没有执行fianlly的释放锁的操作，那么其他的线程无法获取锁。

3.2.3、设置锁的过期时间

/*** 处理锁无法释放的操作* @return*/public String sale() {String retMessage = "";String key = "RedisLock";String uuidValue = IdUtil.simpleUUID() + ":" + Thread.currentThread().getId();
​
​//不适用递归而是使用自璇的方式,改为whilewhile (!stringRedisTemplate.opsForValue().setIfAbsent(key, uuidValue,30L,TimeUnit.SECONDS)) {//没有获取
​//获取锁失败重试try {TimeUnit.MICROSECONDS.sleep(20);}catch (InterruptedException e){e.printStackTrace();}}//抢锁成功!设置锁的过期时间try{//1 查询库存信息String result = stringRedisTemplate.opsForValue().get("inventory001");//2 判断库存是否足够Integer inventoryNumber = result == null ? 0 : Integer.parseInt(result);//3 扣减库存if(inventoryNumber > 0) {stringRedisTemplate.opsForValue().set("inventory001",String.valueOf(--inventoryNumber));retMessage = "成功卖出一个商品，库存剩余: "+inventoryNumber;System.out.println(retMessage);}else{retMessage = "商品卖完了，o(╥﹏╥)o";}}finally {//释放锁stringRedisTemplate.delete(key);}return retMessage + "\t" + "服务端口号：" + port;
​}

        加锁和设置过期时间应该是同时执行，保证原子性，防止加锁和设置过期之间出现异常，但是随之而来的一个问题是，这个过期时间的设置，太短的话业务还没执行完毕就释放锁啦，这样还是会出现高并发时的问题。其他线程获取锁，此时A线程业务执行完毕那么这时候去释放锁，直接将其他线程设置的锁给删除。

3.2.4、解决锁误删的情况

 /*** 自己的锁自己删除* @return*/public String sale() {String retMessage = "";String key = "RedisLock";String uuidValue = IdUtil.simpleUUID() + ":" + Thread.currentThread().getId();
​
​//不适用递归而是使用自璇的方式,改为whilewhile (!stringRedisTemplate.opsForValue().setIfAbsent(key, uuidValue,30L,TimeUnit.SECONDS)) {//没有获取
​//获取锁失败重试try {TimeUnit.MICROSECONDS.sleep(20);}catch (InterruptedException e){e.printStackTrace();}}//抢锁成功!设置锁的过期时间try{//1 查询库存信息String result = stringRedisTemplate.opsForValue().get("inventory001");//2 判断库存是否足够Integer inventoryNumber = result == null ? 0 : Integer.parseInt(result);//3 扣减库存if(inventoryNumber > 0) {stringRedisTemplate.opsForValue().set("inventory001",String.valueOf(--inventoryNumber));retMessage = "成功卖出一个商品，库存剩余: "+inventoryNumber;System.out.println(retMessage);}else{retMessage = "商品卖完了，o(╥﹏╥)o";}}finally {//释放锁,做一个判断if (stringRedisTemplate.opsForValue().get(key).equalsIgnoreCase(uuidValue)){stringRedisTemplate.delete(key);}}return retMessage + "\t" + "服务端口号：" + port;
​}

        再释放锁的时候做出了判断，那么这时候就可以避免不是自己的锁却误删除的情况！但是判断锁的时候和释放锁的操作不是原子操做，也就是不是一体的再高并发下还是可能出现一系列的问题。

3.2.5、判断和删除使用lua脚本

/*** 使用lua脚本* @return*/public String sale() {String retMessage = "";String key = "RedisLock";String uuidValue = IdUtil.simpleUUID() + ":" + Thread.currentThread().getId();
​
​//不适用递归而是使用自璇的方式,改为whilewhile (!stringRedisTemplate.opsForValue().setIfAbsent(key, uuidValue,30L,TimeUnit.SECONDS)) {//没有获取
​//获取锁失败重试try {TimeUnit.MICROSECONDS.sleep(20);}catch (InterruptedException e){e.printStackTrace();}}//抢锁成功!设置锁的过期时间try{//1 查询库存信息String result = stringRedisTemplate.opsForValue().get("inventory001");//2 判断库存是否足够Integer inventoryNumber = result == null ? 0 : Integer.parseInt(result);//3 扣减库存if(inventoryNumber > 0) {stringRedisTemplate.opsForValue().set("inventory001",String.valueOf(--inventoryNumber));retMessage = "成功卖出一个商品，库存剩余: "+inventoryNumber;System.out.println(retMessage);}else{retMessage = "商品卖完了，o(╥﹏╥)o";}}finally {//使用lua脚本删除锁String script="if redis.call('get',KEYS[1])==ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";stringRedisTemplate.execute(new DefaultRedisScript<>(script, Boolean.class), Arrays.asList(key),uuidValue);}return retMessage + "\t" + "服务端口号：" + port;
​}

        使用lua脚本都可以保证判断key和释放锁的操作保证了原子性，然后这样的话其实这个redis锁就已经可以用生产环境，但是哦还可以继续优化！锁的可重入如何实现？

3.2.6、可重入锁

        就是一个线程中的多个流程可以获取同一把锁，持有这个同步锁可以再次进入，自己获取自己的内部锁。

        首先可重入锁可以分为两类，隐式锁是指（syncheonized关键字使用的锁），默认时可以重入的。

同步代码块

public class ReEntryLockDemo
{public static void main(String[] args){final Object objectLockA = new Object();
​new Thread(() -> {synchronized (objectLockA){System.out.println("-----外层调用");synchronized (objectLockA){System.out.println("-----中层调用");synchronized (objectLockA){System.out.println("-----内层调用");}}}},"a").start();}
}

同步方法：

/*** 在一个Synchronized修饰的方法或代码块的内部调用本类的其他Synchronized修饰的方法或代码块时，是永远可以得到锁的*/
public class ReEntryLockDemo
{public synchronized void m1(){System.out.println("-----m1");m2();}public synchronized void m2(){System.out.println("-----m2");m3();}public synchronized void m3(){System.out.println("-----m3");}
​public static void main(String[] args){ReEntryLockDemo reEntryLockDemo = new ReEntryLockDemo();
​reEntryLockDemo.m1();}
}

        显式锁（即Lock）也有ReentrantLock这样的可重入锁。

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
​
/*** 在一个Synchronized修饰的方法或代码块的内部调用本类的其他Synchronized修饰的方法或代码块时，是永远可以得到锁的*/
public class ReEntryLockDemo
{static Lock lock = new ReentrantLock();
​public static void main(String[] args){new Thread(() -> {lock.lock();try{System.out.println("----外层调用lock");lock.lock();try{System.out.println("----内层调用lock");}finally {// 这里故意注释，实现加锁次数和释放次数不一样// 由于加锁次数和释放次数不一样，第二个线程始终无法获取到锁，导致一直在等待。lock.unlock(); // 正常情况，加锁几次就要解锁几次}}finally {lock.unlock();}},"a").start();
​new Thread(() -> {lock.lock();try{System.out.println("b thread----外层调用lock");}finally {lock.unlock();}},"b").start();
​}
}

        使用lua脚本来加锁，返回零说明不存在，hset新建当前线程属于自己的锁BY UUID:ThreadID，返回壹说明已经有锁，需进一步判断是不是当前线程自己的 HEXISTS key uuid:ThreadID，返回零说明不是自己的，返回1说明是自己的锁那么可以将重入次数加一

#加锁
if redis.call('exists',KEYS[1]) == 0 or redis.call('hexists',KEYS[1],ARGV[1]) == 1 then redis.call('hincrby',KEYS[1],ARGV[1],1) redis.call('expire',KEYS[1],ARGV[2]) return 1 
elsereturn 0
end

        使用lua脚本来解锁，首先判断是有锁而且是自己的HEXISTS key uuid:ThreadID，如果返回值是零，那么说明根本没有锁，程序块返回nil，不是零则说明有锁且是自己的锁，直接调用hincrby -1 表示每次减一个，解锁一次直到它变为零表示可以删除该锁，del 锁

if redis.call('HEXISTS',KEYS[1],ARGV[1]) == 0 thenreturn nil
elseif redis.call('HINCRBY',KEYS[1],ARGV[1],-1) == 0 thenreturn redis.call('del',KEYS[1])
elsereturn 0
end

分布式锁工厂

package com.songzhishu.redis.utils;
​
import cn.hutool.core.util.IdUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
​
import java.util.concurrent.locks.Lock;
​
/*** @BelongsProject: redis-lock* @BelongsPackage: com.songzhishu.redis.utils* @Author: 斗痘侠* @CreateTime: 2024-01-21  20:56* @Description: 分步式锁工厂* @Version: 1.0*/
@Component
public class DistributedLockFactory {@Autowiredprivate StringRedisTemplate stringRedisTemplate;private String lockName;
​private String uuid;
​public DistributedLockFactory() {this.uuid = IdUtil.simpleUUID();}
​public Lock getDistributedLock(String lockType){if(lockType == null) return null;
​if(lockType.equalsIgnoreCase("REDIS")){lockName = "RedisLock";return new RedisDistributedLock(stringRedisTemplate,lockName,uuid);} else if(lockType.equalsIgnoreCase("ZOOKEEPER")){//TODO zookeeper版本的分布式锁实现lockName = "ZooKeeperLock";//return new ZookeeperDistributedLock();return null;} else if(lockType.equalsIgnoreCase("MYSQL")){//TODO mysql版本的分布式锁实现return null;}
​return null;}
}
​

​

Redis分布式锁

package com.songzhishu.redis.utils;
​
​
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
​
import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
​
/*** @BelongsProject: redis-lock* @BelongsPackage: com.songzhishu.redis.utils* @Author: 斗痘侠* @CreateTime: 2024-01-21  20:18* @Description: redis的分布式锁* @Version: 1.0*/
public class RedisDistributedLock implements Lock {
​private StringRedisTemplate stringRedisTemplate;
​private String lockName;
​private String uuidValue;
​private long expireTime;public RedisDistributedLock(StringRedisTemplate stringRedisTemplate, String lockName, String uuid) {this.stringRedisTemplate = stringRedisTemplate;this.lockName = lockName;this.uuidValue = uuid + ":" + Thread.currentThread().getId();this.expireTime = 30L;}
​@Overridepublic void lock() {tryLock();}
​
​@Overridepublic boolean tryLock() {try {tryLock(-1L, TimeUnit.SECONDS);} catch (InterruptedException e) {e.printStackTrace();}return false;}
​@Overridepublic boolean tryLock(long time, TimeUnit unit) throws InterruptedException {//1.获取锁
​if (time == -1L) {//使用lua脚本使用hash结构String scirpt = "if redis.call('exists',KEYS[1]) == 0 or redis.call('hexists',KEYS[1],ARGV[1]) == 1 then " +"  redis.call('hincrby',KEYS[1],ARGV[1],1) " +"  redis.call('expire',KEYS[1],ARGV[2]) " +"  return 1 " +"else " +"  return 0 " +"end ";
​
​while (!stringRedisTemplate.execute(new DefaultRedisScript<>(scirpt, Boolean.class), Arrays.asList(lockName), uuidValue, String.valueOf(expireTime))) {//获取锁失败，休眠一段时间try {TimeUnit.MICROSECONDS.sleep(20);} catch (InterruptedException e) {e.printStackTrace();}}return true;}return false;}
​@Overridepublic void unlock() {//使用lua脚本删除锁String script = "if redis.call('HEXISTS',KEYS[1],ARGV[1]) == 0 then " +" return nil " +"elseif redis.call('HINCRBY',KEYS[1],ARGV[1],-1) == 0 then " +" return redis.call('del',KEYS[1]) " +"else " +" return 0 " +"end ";//nil==false,0==false,1==true
​Long flag = stringRedisTemplate.execute(new DefaultRedisScript<>(script, Long.class), Arrays.asList(lockName), uuidValue);if (flag == null) {throw new RuntimeException("解锁异常");}
​}
​@Overridepublic Condition newCondition() {return null;}
​@Overridepublic void lockInterruptibly() throws InterruptedException {
​}
}

3.2.7、自动续期

        自动续期是用来应对由于redislock锁设置过期时间小于业务的执行时间，那么就会导致业务未执行完就释放锁的问题。

package com.songzhishu.redis.utils;
​
​
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
​
import java.util.Arrays;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
​
/*** @BelongsProject: redis-lock* @BelongsPackage: com.songzhishu.redis.utils* @Author: 斗痘侠* @CreateTime: 2024-01-21  20:18* @Description: redis的分布式锁* @Version: 1.0*/
public class RedisDistributedLock implements Lock {
​private StringRedisTemplate stringRedisTemplate;
​private String lockName;
​private String uuidValue;
​private long expireTime;
​public RedisDistributedLock(StringRedisTemplate stringRedisTemplate, String lockName, String uuid) {this.stringRedisTemplate = stringRedisTemplate;this.lockName = lockName;this.uuidValue = uuid + ":" + Thread.currentThread().getId();this.expireTime = 30L;}
​@Overridepublic void lock() {tryLock();}
​
​@Overridepublic boolean tryLock() {try {tryLock(-1L, TimeUnit.SECONDS);} catch (InterruptedException e) {e.printStackTrace();}return false;}
​@Overridepublic boolean tryLock(long time, TimeUnit unit) throws InterruptedException {//1.获取锁
​if (time == -1L) {//使用lua脚本使用hash结构String scirpt = "if redis.call('exists',KEYS[1]) == 0 or redis.call('hexists',KEYS[1],ARGV[1]) == 1 then " +"  redis.call('hincrby',KEYS[1],ARGV[1],1) " +"  redis.call('expire',KEYS[1],ARGV[2]) " +"  return 1 " +"else " +"  return 0 " +"end ";
​
​while (!stringRedisTemplate.execute(new DefaultRedisScript<>(scirpt, Boolean.class), Arrays.asList(lockName), uuidValue, String.valueOf(expireTime))) {//获取锁失败，休眠一段时间try {TimeUnit.MICROSECONDS.sleep(20);} catch (InterruptedException e) {e.printStackTrace();}}
​//获取锁成功,防止业务代码执行时间过长，导致锁过期，其他线程获取到锁renewExpire();return true;}return false;}
​private void renewExpire() {String script = "if redis.call('HEXISTS',KEYS[1],ARGV[1]) == 1 then " +"  return redis.call('expire',KEYS[1],ARGV[2]) " +"else " +"  return 0 " +"end";
​new Timer().schedule(new TimerTask() {@Overridepublic void run() {if (stringRedisTemplate.execute(new DefaultRedisScript<>(script, Boolean.class), Arrays.asList(lockName), uuidValue, String.valueOf(expireTime))) {renewExpire();}}}, (this.expireTime * 1000) / 3);}
​@Overridepublic void unlock() {//使用lua脚本删除锁String script = "if redis.call('HEXISTS',KEYS[1],ARGV[1]) == 0 then " +" return nil " +"elseif redis.call('HINCRBY',KEYS[1],ARGV[1],-1) == 0 then " +" return redis.call('del',KEYS[1]) " +"else " +" return 0 " +"end ";//nil==false,0==false,1==true
​Long flag = stringRedisTemplate.execute(new DefaultRedisScript<>(script, Long.class), Arrays.asList(lockName), uuidValue);if (flag == null) {throw new RuntimeException("解锁异常");}
​}
​@Overridepublic Condition newCondition() {return null;}
​@Overridepublic void lockInterruptibly() throws InterruptedException {
​}
}

4、Redlock算法

        是这样的之前写的使用redis分布式锁，其实上还有一些问题，比如一把锁被两个线程同时持有，在生产中这是严重的问题。

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

        但是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 个节点，官方建议是 5。

容错公式：N = 2X + 1 (N是最终部署机器数，X是容错机器数)

        先知道什么是容错，失败了多少个机器实例后我还是可以容忍的，所谓的容忍就是数据一致性还是可以Ok的，CP数据一致性还是可以满足

• 加入在集群环境中，redis失败1台，可接受。2X+1 = 2 * 1+1 =3，部署3台，死了1个剩下2个可以正常工作，那就部署3台。

• 加入在集群环境中，redis失败2台，可接受。2X+1 = 2 * 2+1 =5，部署5台，死了2个剩下3个可以正常工作，那就部署5台。

为什么是奇数，最少的机器，最多的产出效果

• 加入在集群环境中，redis失败1台，可接受。2N+2= 2 * 1+2 =4，部署4台

• 加入在集群环境中，redis失败2台，可接受。2N+2 = 2 * 2+2 =6，部署6台

4.1、RedisSon

        使用的话也很方便，导入依赖然后写配置类，对于配置类的话可以是单也可以是单机的方式来配置。

依赖

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

配置类

package com.songzhishu.redis.config;
​
​
import org.redisson.Redisson;
import org.redisson.config.Config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
​
​
​
​
/*** @BelongsProject: redis-lock* @BelongsPackage: com.songzhishu.redis.config* @Author: 斗痘侠* @CreateTime: 2024-01-20  21:27* @Description: redis配置类,序列化方式，redisson配置* @Version: 1.0*/
@Configuration
public class RedisConfig
{@Beanpublic RedisTemplate<String, Object> redisTemplate(LettuceConnectionFactory lettuceConnectionFactory){RedisTemplate<String,Object> redisTemplate = new RedisTemplate<>();redisTemplate.setConnectionFactory(lettuceConnectionFactory);//设置key序列化方式stringredisTemplate.setKeySerializer(new StringRedisSerializer());//设置value的序列化方式jsonredisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());
​redisTemplate.setHashKeySerializer(new StringRedisSerializer());redisTemplate.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());
​redisTemplate.afterPropertiesSet();
​return redisTemplate;}
​//单Redis节点模式@Beanpublic Redisson redisson(){Config config = new Config();config.useSingleServer().setAddress("redis://192.168.200.99:6379").setDatabase(0).setPassword("10100109");return (Redisson) Redisson.create(config);}
}

        使用的话就是注入bean，然后调用api就可以，也就是加锁和释放锁，但是直接释放锁的话还是会出现之前的问题，就是不是自己的话锁不能释放，要先判断当前的锁是存在的并且是属于当前的线程的，然后才可以释放锁。

​

    #注入bean @Resourceprivate Redisson redisson;
​
/*** 使用redisson,判断锁是否是当前线程持有* @return*/public String saleByRedisson() {String retMessage = "";RLock redissonLock = redisson.getLock("RedisLock");redissonLock.lock();try
​{//1 查询库存信息String result = stringRedisTemplate.opsForValue().get("inventory001");//2 判断库存是否足够Integer inventoryNumber = result == null ? 0 : Integer.parseInt(result);//3 扣减库存if(inventoryNumber > 0) {stringRedisTemplate.opsForValue().set("inventory001",String.valueOf(--inventoryNumber));retMessage = "成功卖出一个商品，库存剩余: "+inventoryNumber;System.out.println(retMessage);}else{retMessage = "商品卖完了，o(╥﹏╥)o";}}finally {
​//释放锁没有判断，直接释放会出现异常,判断锁是否是当前线程持有并且锁没有过期if (redissonLock.isLocked()&&redissonLock.isHeldByCurrentThread()){redissonLock.unlock();}}return retMessage+"\t"+"服务端口号："+port;}