多執行緒是為了更好的利用 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;

    /**
     * Business logic is not written wrong, for small and medium factories(QPS《=1000)it can be used, but for big factories it is not acceptable
     * @param id
     * @return
     */
    public User findUserById(Integer id)
    {
        User user = null;
        String key = CACHE_KEY_USER+id;

        //1 First from redis query inside, if there is a result return it directly, if not then query mysql
        user = (User) redisTemplate.opsForValue().get(key);

        if(user == null)
        {
            //2 redis nothing inside, continue querying mysql
            user = userMapper.selectByPrimaryKey(id);
            if(user == null)
            {
                //3.1 redis+mysql no data in both
                //you refine it specifically, to prevent repeated penetration, our business rule is, record the one that causes penetration key write back redis
                return user;
            }else{
                //3.2 mysql there is, need to write the data back to redis，ensure the cache hit rate next time
                redisTemplate.opsForValue().set(key,user);
            }
        }
        return user;
    }


    /**
     * strengthen the supplement, avoid sudden key expiration, overwhelming mysql，do some prevention, try not to have breakdown situations.
     * @param id
     * @return
     */
    public User findUserById2(Integer id)
    {
        User user = null;
        String key = CACHE_KEY_USER+id;

        //1 First from redis query inside, if there is a result return it directly, if not then query mysql，
        // Number 1 query time redis，before locking
        user = (User) redisTemplate.opsForValue().get(key);
        if(user == null) {
            //2 used by big factories, for high QPS optimization, lock first when entering, ensure one request operation, let the outside redis wait a bit, avoid breakdown mysql
            synchronized (UserService.class){
                //Number 2 query time redis，after locking
                user = (User) redisTemplate.opsForValue().get(key);
                //3 second check redis still null，can query the mysql it is done(mysql default has data)
                if (user == null) {
                    //4 query mysql get data(mysql default has data)
                    user = userMapper.selectByPrimaryKey(id);
                    if (user == null) {
                        return null;
                    }else{
                        //5 mysql there is data inside, need to write back redis，complete the synchronization work of data consistency
                        redisTemplate.opsForValue().setIfAbsent(key,user,7L,TimeUnit.DAYS);
                    }
                }
            }
        }
        return user;
    }
}

不推薦，業務上把 MySQL 作為底單資料，保證最後解釋

同樣多執行緒

先刪除了reids資料，讓然後mysql進行了更新，mysql沒更新完的時候就存取了資料，存取取得了舊值，還會把舊值寫回了redis，之前就白刪了，就導致新更新的資料沒有同步到redis,redis中資料一直是髒的，採用延時雙刪來解決，也就是在mysql更新後再刪除一次 Redis，延遲刪除時間需要估算

解決方案 : 延時雙刪策略

更新 MySQL 前後兩次進行刪除，負責刪除的 A 執行緒 sleep 時間要大於 B 執行緒寫入 Redis 快取的時間，防止別的執行緒讀取到舊資料寫入 Redis 後沒有被刪除（第二次刪除：將刪除變成非同步的避免吞吐量降低）

• 刪除該休眠多久

• 同步淘汰策略，吞吐量降低怎麼辦

將刪除變成非同步的避免吞吐量降低

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. 下載 deployer 版本然後解壓縮到/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 {
                //get changed row data
                rowChage = RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error,data:" + entry.toString(),e);
            }
            //get change type
            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方法-----------");

        //=================================
        // create connection canal server side
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(REDIS_IP_ADDR,
                11111), "example", "", "");
        int batchSize = 1000;
        //idle spin counter
        int emptyCount = 0;
        System.out.println("---------------------canal init OK，开始监听mysql变化------");
        try {
            connector.connect();
            //connector.subscribe(".*\\..*");
            //more monitoring,slower write back,only guarantee double write consistency for important data
            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); // get specified amount of data
                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 {
                    //reset counter to zero
                    emptyCount = 0;
                    printEntry(message.getEntries());
                }
                connector.ack(batchId); // submit confirmation
                // connector.rollback(batchId); // processing failed, roll back data
            }
            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) {  
    //get longitude and latitude coordinates  
    List<Point> list = redisTemplate.opsForGeo().position(CITY, member);  
    return list.get(0);  
}  
  
public String hash(String member) {  
    //geohash generated by algorithm base32 encoded value  
    List<String> list = redisTemplate.opsForGeo().hash(CITY, member);  
    return list.get(0);  
}  
  
public Distance distance(String member1, String member2) {  
    //get distance between two given locations  
    Distance distance = redisTemplate.opsForGeo().distance(CITY, member1, member2,  
            RedisGeoCommands.DistanceUnit.KILOMETERS);  
    return distance;  
}  
public GeoResults radiusByxy() {  
    //find nearby places by longitude and latitude,Beijing Wangfujing location 116.418017,39.914402  
    //Metrics.KILOMETERS.getMultiplier()radius of circle in kilometers
    Circle circle = new Circle(116.418017, 39.914402, Metrics.KILOMETERS.getMultiplier());  
    // return 50 items  
    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() {  
    //find nearby places by location  
    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 誤判率，它越小誤判的個數也就越少(Thinking, can it be infinitely small? Would it not be better without false positives?)
    public static double fpp = 0.01;//0.01 0.000000000000001
    //4 Create guava Bloom filter
    private static BloomFilter<Integer> bloomFilter = BloomFilter.create(Funnels.integerFunnel(), SIZE,fpp);


    public void guavaBloomFilter()
    {
        //1 First let bloomFilter add 100W whitelist data
        for (int i = 1; i <= SIZE  ; i++) {
            bloomFilter.put(i);
        }
        //2 Deliberately take 10W pieces of data outside the valid range,to demonstrate the false positive rate
        ArrayList<Integer> list = new ArrayList<>(10 * _1W);

        //3 Verify
        //The filter only added SIZE data,Deliberately start from SIZE+1 data outside to begin judging,false positives occur
        for (int i = SIZE+1; i <= SIZE+(10 * _1W) ; i++)
        {
            if(bloomFilter.mightContain(i))
            {
                log.info("被误判了:{}",i);
                list.add(i);
            }
        }
        log.info("误判总数量:{}",list.size());
    }
}

For example, a Double Eleven business only runs for a few days, after that time it is cleared but then accessed again Right after deletion, other threads may access it again

原理

同步區塊

同步方法

為了 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