多线程是为了更好的利用 CPU，而对于 redis 来说主要瓶颈是内存或者网络带宽而非 CPU，而且即使单线程也能通过多路复用和非阻塞 IO 来并发处理客户端的请求, 最初单线程的维护和开发更加简单

例如时包含了成千上万个元素的 hash 集合时，那么 del 指令就会造成 Redis 主线程卡顿

等待很久这个线程才会释放，类似加了一个synchronized锁，你可以想象高并发下，程序会超级堵

一般来说CPU,内存都不是问题,网络IO是主要的问题

虽然有些命令操作可以用后台线程或子进程执行（比如数据删除、快照生成、AOF重写）。但是，从网络IO处理到实际的读写命令处理，都是由单个线程完成的。Redis的性能瓶颈有时会出现在网络IO的处理上，也就是说，单个主线程处理网络请求的速度跟不上底层网络硬件的速度,为了应对这个问题:

采用多个IO线程来处理网络请求，提高网络请求处理的并行度，Redis6/7就是采用的这种方法

也就是网络IO是多线程,真正干活的线程还是单线程

1. 连接,建立了Socket连接

IO线程读取和解析,比如网络是否通,IP是否允许,防火墙是否允许

文件描述符 FD 句柄

IO多路复用是什么

"文件句柄就绪"通常指的是操作系统中的文件描述符（file descriptor）或文件句柄（file handle）处于一种可以进行I/O操作的状态。

1.设置io-thread-do-reads配置项为yes，表示启动多线程。

2。设置线程个数。关于线程数的设置，官方的建议是如果为 4 核的 CPU，建议线程数设置为 2 或 3，如果为 8 核 CPU 建议线程数设置为 6，线程数一定要小于机器核数，线程数并不是越大越好。

线程不是越大越好

直接配置文件中配置成空串 ""

《阿里云规范》

1. redis-cli xxxx 结尾加上 --bigkeys 虽然--bigkeys命令可以帮助你找到一些大键，但是当键的大小超过一定限制时，它可能无法提供高效的性能，甚至可能会对Redis实例的稳定性和整体性能产生负面影响。

2. memory usage key

从redis获取,redis无则从mysql取

@Service
@Slf4j
public class UserService {
    public static final String CACHE_KEY_USER = "user:";
    @Resource
    private UserMapper userMapper;
    @Resource
    private RedisTemplate redisTemplate;

    /**
     * 业务逻辑没有写错，对于小厂中厂(QPS《=1000)可以使用，但是大厂不行
     * @param id
     * @return
     */
    public User findUserById(Integer id)
    {
        User user = null;
        String key = CACHE_KEY_USER+id;

        //1 先从redis里面查询，如果有直接返回结果，如果没有再去查询mysql
        user = (User) redisTemplate.opsForValue().get(key);

        if(user == null)
        {
            //2 redis里面无，继续查询mysql
            user = userMapper.selectByPrimaryKey(id);
            if(user == null)
            {
                //3.1 redis+mysql 都无数据
                //你具体细化，防止多次穿透，我们业务规定，记录下导致穿透的这个key回写redis
                return user;
            }else{
                //3.2 mysql有，需要将数据写回redis，保证下一次的缓存命中率
                redisTemplate.opsForValue().set(key,user);
            }
        }
        return user;
    }


    /**
     * 加强补充，避免突然key失效了，打爆mysql，做一下预防，尽量不出现击穿的情况。
     * @param id
     * @return
     */
    public User findUserById2(Integer id)
    {
        User user = null;
        String key = CACHE_KEY_USER+id;

        //1 先从redis里面查询，如果有直接返回结果，如果没有再去查询mysql，
        // 第1次查询redis，加锁前
        user = (User) redisTemplate.opsForValue().get(key);
        if(user == null) {
            //2 大厂用，对于高QPS的优化，进来就先加锁，保证一个请求操作，让外面的redis等待一下，避免击穿mysql
            synchronized (UserService.class){
                //第2次查询redis，加锁后
                user = (User) redisTemplate.opsForValue().get(key);
                //3 二次查redis还是null，可以去查mysql了(mysql默认有数据)
                if (user == null) {
                    //4 查询mysql拿数据(mysql默认有数据)
                    user = userMapper.selectByPrimaryKey(id);
                    if (user == null) {
                        return null;
                    }else{
                        //5 mysql里面有数据的，需要回写redis，完成数据一致性的同步工作
                        redisTemplate.opsForValue().setIfAbsent(key,user,7L,TimeUnit.DAYS);
                    }
                }
            }
        }
        return user;
    }
}

不推荐,业务上把mysql作为底单数据,保证最后解释

同样多线程

先删除了reids数据,让然后mysql进行了更新,mysql没更新完的时候就访问了数据,访问获取了旧值,还会把旧值写回了redis,之前就白删了,就导致新更新的数据没有同步到redis,redis中数据一直是脏的,采用延时双删来解决,也就是在mysql更新后再删除一次redis,延迟删除时间需要估算

解决方案 : 延时双删策略

更新mysql前后两次进行删除,负责删除的A线程sleep时间要大于B线程写入redis缓存的时间,防止别的线程读取到旧数据写入reids后没有被删除(第二次删除:将删除变成异步的避免吞吐量降低)

• 删除该休眠多久

• 同步淘汰策略,吞吐量降低怎么办法

将删除变成异步的避免吞吐量降低

1. 异常情况: 多线程下,顺序错乱 1.异常情况:redis更新失败,redis读到脏数据

问题还是读到旧值,但mysql更新完后旧值就会被替换,下次的请求就会得到最新的值,不会引发其他新的问题

但如果删除失败就有问题了

• mysql配置修改

开启mysql的binlog写入功能

之后会变成ON

新建canal账户并授权 select * from mysql.user

DROP USER IF EXISTS 'canal'@'%'; CREATE USER 'canal'@'%' IDENTIFIED BY 'canal';   GRANT ALL PRIVILEGES ON . TO 'canal'@'%' IDENTIFIED BY 'canal';   FLUSH PRIVILEGES; SELECT * FROM mysql.user;

1. 下载depolyer版本然后解压到/mycanal路径下
2. 配置 将master填写自己的mysql地址 将账户换成之前自己新建的

<dependency>
    <groupId>com.alibaba.otter</groupId>
    <artifactId>canal.client</artifactId>
    <version>1.1.0</version>
</dependency>


<dependency>
		<groupId>redis.clients</groupId>
		<artifactId>jedis</artifactId>
		<version>3.8.0</version>
</dependency>

public class RedisUtils
{
    public static final String  REDIS_IP_ADDR = "192.168.111.185";
    public static final String  REDIS_pwd = "111111";
    public static JedisPool jedisPool;

    static {
        JedisPoolConfig jedisPoolConfig=new JedisPoolConfig();
        jedisPoolConfig.setMaxTotal(20);
        jedisPoolConfig.setMaxIdle(10);
        jedisPool=new JedisPool(jedisPoolConfig,REDIS_IP_ADDR,6379,10000,REDIS_pwd);
    }

    public static Jedis getJedis() throws Exception {
        if(null!=jedisPool){
            return jedisPool.getResource();
        }
        throw new Exception("Jedispool is not ok");
    }

}

public class RedisCanalClientExample
{
    public static final Integer _60SECONDS = 60;
    public static final String  REDIS_IP_ADDR = "192.168.111.185";

    private static void redisInsert(List<Column> columns)
    {
        JSONObject jsonObject = new JSONObject();
        for (Column column : columns)
        {
            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
            jsonObject.put(column.getName(),column.getValue());
        }
        if(columns.size() > 0)
        {
            try(Jedis jedis = RedisUtils.getJedis())
            {
                jedis.set(columns.get(0).getValue(),jsonObject.toJSONString());
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }


    private static void redisDelete(List<Column> columns)
    {
        JSONObject jsonObject = new JSONObject();
        for (Column column : columns)
        {
            jsonObject.put(column.getName(),column.getValue());
        }
        if(columns.size() > 0)
        {
            try(Jedis jedis = RedisUtils.getJedis())
            {
                jedis.del(columns.get(0).getValue());
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }

    private static void redisUpdate(List<Column> columns)
    {
        JSONObject jsonObject = new JSONObject();
        for (Column column : columns)
        {
            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
            jsonObject.put(column.getName(),column.getValue());
        }
        if(columns.size() > 0)
        {
            try(Jedis jedis = RedisUtils.getJedis())
            {
                jedis.set(columns.get(0).getValue(),jsonObject.toJSONString());
                System.out.println("---------update after: "+jedis.get(columns.get(0).getValue()));
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }

    public static void printEntry(List<Entry> entrys) {
        for (Entry entry : entrys) {
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                continue;
            }

            RowChange rowChage = null;
            try {
                //获取变更的row数据
                rowChage = RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error,data:" + entry.toString(),e);
            }
            //获取变动类型
            EventType eventType = rowChage.getEventType();
            System.out.println(String.format("================&gt; binlog[%s:%s] , name[%s,%s] , eventType : %s",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), eventType));

            for (RowData rowData : rowChage.getRowDatasList()) {
                if (eventType == EventType.INSERT) {
                    redisInsert(rowData.getAfterColumnsList());
                } else if (eventType == EventType.DELETE) {
                    redisDelete(rowData.getBeforeColumnsList());
                } else {//EventType.UPDATE
                    redisUpdate(rowData.getAfterColumnsList());
                }
            }
        }
    }


    public static void main(String[] args)
    {
        System.out.println("---------O(∩_∩)O哈哈~ initCanal() main方法-----------");

        //=================================
        // 创建链接canal服务端
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(REDIS_IP_ADDR,
                11111), "example", "", "");
        int batchSize = 1000;
        //空闲空转计数器
        int emptyCount = 0;
        System.out.println("---------------------canal init OK，开始监听mysql变化------");
        try {
            connector.connect();
            //connector.subscribe(".*\\..*");
            //监控越多,回写越慢,只对重要数据保证双写一致性
            connector.subscribe("bigdata.t_user");
            connector.rollback();
            int totalEmptyCount = 10 * _60SECONDS;
            while (emptyCount < totalEmptyCount) {
                System.out.println("我是canal，每秒一次正在监听:"+ UUID.randomUUID().toString());
                Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    emptyCount++;
                    try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); }
                } else {
                    //计数器重新置零
                    emptyCount = 0;
                    printEntry(message.getEntries());
                }
                connector.ack(batchId); // 提交确认
                // connector.rollback(batchId); // 处理失败, 回滚数据
            }
            System.out.println("已经监听了"+totalEmptyCount+"秒，无任何消息，请重启重试......");
        } finally {
            connector.disconnect();
        }
    }
}

java程序下connector.subscribe配置的过滤正则

try-with-resources无需再手动关闭

jedis会自动关闭,所以不需要手动finally

由于bitmap在数据量过大时占用了过多内存,因此演化出了hyperloglog,内存减少的效果显著

http://antirez.com/news/75

1.04/16384/16384 = 0.0081

public static final String CITY ="city";  

@Autowired  
private RedisTemplate redisTemplate;
public String geoAdd()  
{  
    Map<String, Point> map = new HashMap<>();  
    map.put("天安门",new Point(116.403963,39.915119));  
    map.put("故宫",new Point(116.403414,39.924091));  
    map.put("长城",new Point(116.024067,40.362639));  
  
    redisTemplate.opsForGeo().add(CITY,map);  
  
    return map.toString();  
}  
  
public Point position(String member) {  
    //获取经纬度坐标  
    List<Point> list = redisTemplate.opsForGeo().position(CITY, member);  
    return list.get(0);  
}  
  
public String hash(String member) {  
    //geohash算法生成的base32编码值  
    List<String> list = redisTemplate.opsForGeo().hash(CITY, member);  
    return list.get(0);  
}  
  
public Distance distance(String member1, String member2) {  
    //获取两个给定位置之间的距离  
    Distance distance = redisTemplate.opsForGeo().distance(CITY, member1, member2,  
            RedisGeoCommands.DistanceUnit.KILOMETERS);  
    return distance;  
}  
public GeoResults radiusByxy() {  
    //通过经度，纬度查找附近的,北京王府井位置116.418017,39.914402  
    //Metrics.KILOMETERS.getMultiplier()以千米为单位的圆的半径
    Circle circle = new Circle(116.418017, 39.914402, Metrics.KILOMETERS.getMultiplier());  
    // 返回50条  
    RedisGeoCommands.GeoRadiusCommandArgs args = RedisGeoCommands.GeoRadiusCommandArgs.newGeoRadiusArgs().includeDistance().includeCoordinates().sortDescending().limit(50);  
  
    GeoResults<RedisGeoCommands.GeoLocation<String>> geoResults = redisTemplate.opsForGeo().radius(CITY, circle, args);  
  
    return geoResults;  
}  
public GeoResults radiusByMember() {  
    //通过地方查找附近  
    return null;  
}

可以用多个坑位联合表示一个数据是否存在,多个坑位同时存在才存在

所以布隆过滤器添加完成后尽量减少删除

只有一个hash函数时,容易发生哈希碰撞,多个hash函数能减少哈希碰撞

• 初始化bitmap

• 添加占坑位

• 判断是否存在

计算哈希后通过hashValue和2的32次方取余后得到对应的下标坑位

1. 通过BloomFlterInit将已经存在数据作为白名单加入到布隆过滤器

2. 通过CheckUtils检查新的数据是否在布隆过滤器

3. 得到哈希值 int hashvalues = Math.abs(key.hashCode());

4. 得到槽数 long v = (long) (hashvalues % Math.pow(2, 32));

5. 设置bit redisTemplate.opsForValue().setBit("whitelistCustomer",v, true);

* 布隆过滤器白名单初始化工具类，一开始就设置一部分数据为白名单所有，  
* 白名单业务默认规定：布隆过滤器有，redis是极大可能有。


@Component
public class BloomFilterInit  
{  
    @Resource  
    private RedisTemplate redisTemplate;  
  
    @PostConstruct//初始化白名单数据,暂时注释省的后台打印  
    public void init()  
    {  
        //1 白名单客户加载到布隆过滤器  
        String key = "customer:12";  
        //2 计算hashValue,由于存在计算出来负数的可能，我们取绝对值  
        int hashValue = Math.abs(key.hashCode());  
        //3 通过hashValue和2的32次方后取余，获得对应的下标坑位  
        long index = (long)(hashValue % Math.pow(2,32));  
        log.info(key+" 对应的坑位index:{}",index);  
        //4 设置redis里面的bitmap对应类型白名单：whitelistCustomer的坑位，将该值设置为1  
        redisTemplate.opsForValue().setBit("whitelistCustomer",index,true);  
  
    }  
}

public class CheckUtils  
{  
    @Resource  
    private RedisTemplate redisTemplate;  
  
    public boolean checkWithBloomFilter(String checkItem,String key)  
    {  
        int hashValue = Math.abs(key.hashCode());  
        long index = (long)(hashValue % Math.pow(2,32)); 
        ////checkItem就是key值 
        boolean existOK = redisTemplate.opsForValue().getBit(checkItem,index);  
        log.info("--->key:"+key+" 对应坑位下标index: "+index+" 是否存在："+existOK);  
  
        return existOK;  
    }  
}

public Customer findCustomerByIdWithBloomFilter (Integer customerId)  
{  
    Customer customer = null;  
    //缓存key的名称  
    String key = CACHA_KEY_CUSTOMER + customerId;  
  
    //布隆过滤器check，无是绝对无，有是可能有  
    //===============================================  
    if(!checkUtils.checkWithBloomFilter("whitelistCustomer",key))  
    {  
        log.info("白名单无此顾客，不可以访问: "+key);  
        return null;    }  
    //===============================================  
  
    //1 查询redis  
    customer = (Customer) redisTemplate.opsForValue().get(key);  
    //redis无，进一步查询mysql  
    if (customer == null) {  
        //2 从mysql查出来customer  
        customer = customerMapper.selectByPrimaryKey(customerId);  
        // mysql有，redis无  
        if (customer != null) {  
            //3 把mysql捞到的数据写入redis，方便下次查询能redis命中。  
            redisTemplate.opsForValue().set(key, customer);  
        }  
    }  
    return customer;  
}

误判率越小：生成的槽位越多，使用的哈希函数越多

但是，增加哈希函数的数量也会增加计算开销。通常情况下，合理的哈希函数数量可以在保持低误判率的同时降低计算成本

这10万数据都不在布隆过滤器中,应该都不存在,但由于过滤器误判可能判为存在

@Service
@Slf4j
public class GuavaBloomFilterService
{
    //1 定义一个常量
    public static final int _1W = 10000;
    //2 定义我们guava布隆过滤器，初始容量
    public static final int SIZE = 100 * _1W;
    //3 误判率，它越小误判的个数也就越少(思考，是否可以是无限小？？没有误判岂不是更好)
    public static double fpp = 0.01;//0.01 0.000000000000001
    //4 创建guava布隆过滤器
    private static BloomFilter<Integer> bloomFilter = BloomFilter.create(Funnels.integerFunnel(), SIZE,fpp);


    public void guavaBloomFilter()
    {
        //1 先让bloomFilter加入100W白名单数据
        for (int i = 1; i <= SIZE  ; i++) {
            bloomFilter.put(i);
        }
        //2 故意取10W个不在合法范围内的数据,来进行误判率的演示
        ArrayList<Integer> list = new ArrayList<>(10 * _1W);

        //3 验证
        //过滤器只添加了SIZE的数据,故意从SIZE+1外的数据开始判断,出现误判
        for (int i = SIZE+1; i <= SIZE+(10 * _1W) ; i++)
        {
            if(bloomFilter.mightContain(i))
            {
                log.info("被误判了:{}",i);
                list.add(i);
            }
        }
        log.info("误判总数量:{}",list.size());
    }
}

比如双十一的业务只干几天,时间过后清除了但又被访问到了 刚刚删除后可能又其他线程来访问

原理

同步块

同步方法

为了 lock 希望将下面一段抽取出来

希望将这一段抽进一个方法变成lock

• Lock : 加锁并计数的lua脚本

判断锁是否存在，存在创建，计数默认为1，设置过期时间 如果判断存在，且是自己的，说明重用，设置计数器加一，设置过期时间

返回0没有，返回1有

优化为 然后改写为lua脚本

hincrby如果没有会自动创建

unlock : 将这一段抽进一个方法变成unlock

判断锁是否存在且为自己的，然后计数器减一 判断计数器如果为零直接删除 如果不存在直接返回

call方法 return 0 时，它通常表示执行成功并返回一个整数值，就相当于java中return

计数器减到0说明已经全部被解锁，可以删除

@Component  
public class DistributedLockFactory  
{  
    @Autowired  
    private 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")){  
            this.lockName = "zzyyRedisLock";  
            return new RedisDistributedLock(stringRedisTemplate,lockName,uuid);  
        }else if(lockType.equalsIgnoreCase("ZOOKEEPER")){  
            this.lockName = "zzyyZookeeperLockNode";  
            //TODO zookeeper版本的分布式锁  
            return null;  
        }else if(lockType.equalsIgnoreCase("MYSQL")){  
            //TODO MYSQL版本的分布式锁  
            return null;  
        }  
  
        return null;  
    }  
}

lock()会调用 tryLock() tryLock()调用tryLock(long time, TimeUnit unit)

import cn.hutool.core.util.IdUtil;  
import org.springframework.data.redis.core.StringRedisTemplate;  
import org.springframework.data.redis.core.script.DefaultRedisScript;  
import org.springframework.stereotype.Component;  
  
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;  


/**  
 * @auther zzyy  
 * @create 2023-01-09 16:41  
 * 我们自研的redis分布式锁，实现了Lock接口  
 */  
//@Component 引入DistributedLockFactory工厂模式，从工厂获得即可  
public class RedisDistributedLock implements Lock  
{  
    private StringRedisTemplate stringRedisTemplate;  
  
    private String lockName;//KEYS[1]  
    private String uuidValue;//ARGV[1]  
    private long   expireTime;//ARGV[2]  
  
  
    public RedisDistributedLock(StringRedisTemplate stringRedisTemplate, String lockName, String uuid)  
    {  
        this.stringRedisTemplate = stringRedisTemplate;  
        this.lockName = lockName;  
        this.uuidValue = uuid+":"+Thread.currentThread().getId();  
        this.expireTime = 30L;  
    }  
  
  
    @Override  
    public void lock()  
    {  
        tryLock();  
    }  
    @Override  
    public boolean tryLock()  
    {  
        try {tryLock(-1L,TimeUnit.SECONDS);} catch (InterruptedException e) {e.printStackTrace();}  
        return false;  
    }  
  
    @Override  
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException  
    {  
        if(time == -1L)  
        {  
            String script =  
                    "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";  
            System.out.println("lockName:"+lockName+"\t"+"uuidValue:"+uuidValue);  
            //执行脚本为true成功，但while中为false会跳出循环，也就是需要取反  
            while(!stringRedisTemplate.execute(new DefaultRedisScript<>(script,Boolean.class), Arrays.asList(lockName), uuidValue,String.valueOf(expireTime)))  
            {  
                //暂停60毫秒  
                try { TimeUnit.MILLISECONDS.sleep(60); } catch (InterruptedException e) { e.printStackTrace(); }  
            }  
            //新建一个后台扫描程序，来坚持key目前的ttl，是否到我们规定的1/2 1/3来实现续期  
            renewExpire();  
            return true;        }  
        return false;  
    }  
  
  
    //执行解锁脚本  
    @Override  
    public void unlock()  
    {  
        System.out.println("unlock(): lockName:"+lockName+"\t"+"uuidValue:"+uuidValue);  
        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 = null 1 = true 0 = false 返回值有三套值  
        Long flag = stringRedisTemplate.execute(new DefaultRedisScript<>(script, Long.class), Arrays.asList(lockName), uuidValue, String.valueOf(expireTime));  
  
        //如果没锁会返回null,说明超时，自己的锁会自动消失，不会影响其他线程创建自己的锁（key：value，number）  
        if(null == flag)  
        {  
            throw new RuntimeException("this lock doesn't exists，o(╥﹏╥)o");  
        }  
    }  
	  //自动续期
    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";  
        //timer是java的定时器  
        new Timer().schedule(new TimerTask()  
        {  
            @Override  
            public void run()  
            {  
                if (stringRedisTemplate.execute(new DefaultRedisScript<>(script, Boolean.class), Arrays.asList(lockName), uuidValue, String.valueOf(expireTime)))  
                {  
                    renewExpire();  
                }  
            }  
        },(this.expireTime * 1000)/3);  
    }  
  
  
    //====下面两个暂时用不到，不再重写  
    //不需要考虑中断,用完就删就行了  
    @Override  
    public void lockInterruptibly() throws InterruptedException  
    {  
  
    }  
    @Override  
    public Condition newCondition()  
    {  
        return null;  
    }  
}

//V8.0，实现自动续期功能的完善，后台自定义扫描程序，如果规定时间内没有完成业务逻辑，会调用加钟自动续期的脚本  
public String sale()  
{  
    String retMessage = "";  
  
    Lock redisLock = distributedLockFactory.getDistributedLock("redis");  
    redisLock.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+"\t"+"服务端口号"+port);  
            //暂停120秒钟线程,故意的，演示自动续期的功能。。。。。。  
            try { TimeUnit.SECONDS.sleep(120); } catch (InterruptedException e) { e.printStackTrace(); }  
  
        }else{  
            retMessage = "商品卖完了,o(╥﹏╥)o";  
        }  
    }finally {  
        redisLock.unlock();  
    }  
    return retMessage+"\t"+"服务端口号"+port;  
}

第二步就是依次尝试在每个Redis 实例上获取锁，如果超过超时时间就直接下一次redis获取，最后尝试完所有的redis所消耗的时间，减去第一步消耗的时间就是获取锁耗时

容错公式

2x+1

x为挂掉的个数

1. 改pom

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


2. 在configuration中redisConfig中的配置（或者配置yaml）

@Bean  
public Redisson redisson()  
{  
    Config config = new Config();  
    //添加多台,这里只是演示
    config.useSingleServer().setAddress("redis://192.168.111.185:6379").setDatabase(0).setPassword("111111");  
    
  // or read config from file
//config = Config.fromYAML(new File("config-file.yaml")); 

    return (Redisson) Redisson.create(config);  
}

@Autowired  
private StringRedisTemplate stringRedisTemplate;

@Value("${server.port}")  
private String port;

@Autowired
private DistributedLockFactory distributedLockFactory;  

//V9.1,引入Redisson对应的官网推荐RedLock算法实现类
@Autowired  
private Redisson redisson;  
public String saleByRedisson()  
{  
    String retMessage = "";  
  
    RLock redissonLock = redisson.getLock("zzyyRedisLock");  
    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+"\t"+"服务端口号"+port);  
        }else{  
            retMessage = "商品卖完了,o(╥﹏╥)o";  
        }  
    }finally {  
        //改进点，只能删除属于自己的key，不能删除别人的  
        if(redissonLock.isLocked() && redissonLock.isHeldByCurrentThread())  
        {  
            redissonLock.unlock();  
        }  
    }  
    return retMessage+"\t"+"服务端口号"+port;  
}

新new出的默认30秒

加锁过程

scheduleExpriationRenewal给缓存续期

续期

解锁

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


server.port=9090
spring.application.name=redlock


spring.swagger2.enabled=true


spring.redis.database=0
spring.redis.password=
spring.redis.timeout=3000
spring.redis.mode=single
//池化技术
spring.redis.pool.conn-timeout=3000
spring.redis.pool.so-timeout=3000
spring.redis.pool.size=10

spring.redis.single.address1=192.168.111.185:6381
spring.redis.single.address2=192.168.111.185:6382
spring.redis.single.address3=192.168.111.185:6383