Redis為什么速度快

1、完全基于內存，絕大部分請求是純粹的內存操作，非?？焖?。數(shù)據(jù)存在內存中，類似于HashMap，HashMap的優(yōu)勢就是查找和操作的時間復雜度都是O(1)；

2、數(shù)據(jù)結構簡單，對數(shù)據(jù)操作也簡單，Redis中的數(shù)據(jù)結構是專門進行設計的；

3、采用單線程，避免了不必要的上下文切換和競爭條件，也不存在多進程或者多線程導致的切換而消耗 CPU，不用去考慮各種鎖的問題，不存在加鎖釋放鎖操作，沒有因為可能出現(xiàn)死鎖而導致的性能消耗；

4、使用多路I/O復用模型，非阻塞IO；

5、使用底層模型不同，它們之間底層實現(xiàn)方式以及與客戶端之間通信的應用協(xié)議不一樣，Redis直接自己構建了VM 機制 ，因為一般的系統(tǒng)調用系統(tǒng)函數(shù)的話，會浪費一定的時間去移動和請求；

以上幾點都比較好理解，下邊我們針對多路 I/O 復用模型進行簡單的探討：

（1）多路 I/O 復用模型

多路I/O復用模型是利用 select、poll、epoll 可以同時監(jiān)察多個流的 I/O 事件的能力，在空閑的時候，會把當前線程阻塞掉，當有一個或多個流有 I/O 事件時，就從阻塞態(tài)中喚醒，于是程序就會輪詢一遍所有的流（epoll 是只輪詢那些真正發(fā)出了事件的流），并且只依次順序的處理就緒的流，這種做法就避免了大量的無用操作。

這里“多路”指的是多個網絡連接，“復用”指的是復用同一個線程。采用多路 I/O 復用技術可以讓單個線程高效的處理多個連接請求（盡量減少網絡 IO 的時間消耗），且 Redis 在內存中操作數(shù)據(jù)的速度非?？?，也就是說內存內的操作不會成為影響Redis性能的瓶頸，主要由以上幾點造就了 Redis 具有很高的吞吐量。

==select和epoll，poll的區(qū)別==

Redis 有哪些數(shù)據(jù)結構？

• 普通數(shù)據(jù)結構：

1. 字符串 String
2. 字典Hash
3. 列表List
4. 集合Set
5. 有序集合 SortedSet

• 中級數(shù)據(jù)結構：

1. HyperLogLog
2. Geo
3. Pub / Sub

• 高端數(shù)據(jù)結構：

1. BloomFilter
2. RedisSearch
3. Redis-ML
4. JSON

==Redis數(shù)據(jù)結構的應用場景==

有序集合底層原理

• 有序集合對象的編碼可以是ziplist或者skiplist
• ziplist編碼的有序集合對象底層實現(xiàn)是壓縮列表，每個有序集合的元素使用兩個緊挨在一起的壓縮列表節(jié)點來保存，第一個節(jié)點保存有序集合的元素，第二個節(jié)點保存元素的分值。壓縮列表的集合元素按照分值從小到大開始排序，分值較小的節(jié)點靠近壓縮列表的表頭方向，分值較大的節(jié)點靠近壓縮列表的表尾方向
• skiplist編碼的有序集合對象使用zset作為底層實現(xiàn)，一個zset結構同時包含一個字典和一個跳躍表。zset結構中的zsl跳躍表按照分值從小到大保存了所有集合元素，每個跳躍表節(jié)點都保存了一個集合元素，跳躍表節(jié)點的object屬性保存了元素的成員，score屬性保存了元素的分支，通過跳躍表，程序可以對有序集合進行范圍型操作，如ZRANK、ZRANGE等命令。zset結構中的dict字典為有序集合創(chuàng)建了一個從成員到分值的映射，字典的每一個鍵值對都保存了一個集合元素，字典的鍵保存了元素的成員，字典的值保存了元素的分值，通過這個字典，程序可以以O(1)復雜度查找給定成員的分值，ZSCORE命令就是根據(jù)這一特性實現(xiàn)的。

==跳躍表原理==

為什么有序集合需要同時使用跳躍表和字典來實現(xiàn)呢

理論上來講，無論有序集合單獨使用跳躍表和字典來實現(xiàn)有序集合，性能都會比同時使用跳躍表和字典有所降低。例如，如只使用字典來實現(xiàn)有序集合，那么在使用有序集合的ZRANK、ZRANGE命令時，程序都需要對字典保存的所有元素進行排序，完成這種排序的復雜度為O(nlog(n)),以及額外的O(N)的空間（創(chuàng)建數(shù)組保存排序后的元素）；如只使用跳躍表實現(xiàn)有序集合，那么在根據(jù)指定成員查詢分值時，復雜度就會變成O(log(n))，而不是字典的O(1)。綜合以上原因，為了讓有序集合同事?lián)碛凶值浜吞S表的所有特點，Redis選擇同時使用跳躍表和字典來實現(xiàn)有序集合。

skiplist中的zset為什么不使用紅黑數(shù)而使用跳躍表

簡單分析一下skiplist的時間復雜度和空間復雜度，以便對于skiplist的性能有一個直觀的了解。

我們先來計算一下每個節(jié)點所包含的平均指針數(shù)目（概率期望）。節(jié)點包含的指針數(shù)目，相當于這個算法在空間上的額外開銷(overhead)，可以用來度量空間復雜度。

跳躍表中產生越高的節(jié)點層數(shù)，概率越低。定量的分析如下：

• 節(jié)點層數(shù)至少為1。而大于1的節(jié)點層數(shù)，滿足一個概率分布。
• 節(jié)點層數(shù)恰好等于1的概率為1-p。
• 節(jié)點層數(shù)大于等于2的概率為p，而節(jié)點層數(shù)恰好等于2的概率為p(1-p)。
• 節(jié)點層數(shù)大于等于3的概率為p^2，
  而節(jié)點層數(shù)恰好等于3的概率為p^2(1-p)。
• 節(jié)點層數(shù)大于等于4的概率為p^3，
  而節(jié)點層數(shù)恰好等于4的概率為p^3(1-p)。
  因此，一個節(jié)點的平均層數(shù)（也即包含的平均指針數(shù)目），計算如下：

(1-p)+2p(1-p)+3p^2 (1-p)+4p^3 (1-)p+kp^(k-1) (1-p)=(1-)p·1/(1-p)^2=1/(1-p)

上述結果可以使用錯位相減法計算得到。

現(xiàn)在很容易計算出：

• 當p=1/2時，每個節(jié)點所包含的平均指針數(shù)目為2；
• 當p=1/4時，每個節(jié)點所包含的平均指針數(shù)目為1.33。這也是Redis里的skiplist實現(xiàn)在空間上的開銷。

==skiplist與平衡樹、哈希表的比較==

• skiplist和各種平衡樹（如AVL、紅黑樹等）的元素是有序排列的，而哈希表不是有序的。因此，在哈希表上只能做單個key的查找，不適宜做范圍查找。所謂范圍查找，指的是查找那些大小在指定的兩個值之間的所有節(jié)點。
• 在做范圍查找的時候，平衡樹比skiplist操作要復雜。在平衡樹上，我們找到指定范圍的小值之后，還需要以中序遍歷的順序繼續(xù)尋找其它不超過大值的節(jié)點。如果不對平衡樹進行一定的改造，這里的中序遍歷并不容易實現(xiàn)。而在skiplist上進行范圍查找就非常簡單，只需要在找到小值之后，對第1層鏈表進行若干步的遍歷就可以實現(xiàn)。
• 平衡樹的插入和刪除操作可能引發(fā)子樹的調整，邏輯復雜，而skiplist的插入和刪除只需要修改相鄰節(jié)點的指針，操作簡單又快速。
• 從內存占用上來說，skiplist比平衡樹更靈活一些。一般來說，平衡樹每個節(jié)點包含2個指針（分別指向左右子樹），而skiplist每個節(jié)點包含的指針數(shù)目平均為1/(1-p)，具體取決于參數(shù)p的大小。如果像Redis里的實現(xiàn)一樣，取p=1/4，那么平均每個節(jié)點包含1.33個指針，比平衡樹更有優(yōu)勢。
• 查找單個key，skiplist和平衡樹的時間復雜度都為O(log n)，大體相當；而哈希表在保持較低的哈希值沖突概率的前提下，查找時間復雜度接近O(1)，性能更高一些。所以我們平常使用的各種Map或dictionary結構，大都是基于哈希表實現(xiàn)的。
• 從算法實現(xiàn)難度上來比較，skiplist比平衡樹要簡單得多。

Redis中的skiplist和經典有何不同

• 分數(shù)(score)允許重復，即skiplist的key允許重復。這在最開始介紹的經典skiplist中是不允許的。
• 在比較時，不僅比較分數(shù)（相當于skiplist的key），還比較數(shù)據(jù)本身。在Redis的skiplist實現(xiàn)中，數(shù)據(jù)本身的內容唯一標識這份數(shù)據(jù)，而不是由key來唯一標識。另外，當多個元素分數(shù)相同的時候，還需要根據(jù)數(shù)據(jù)內容來進字典排序。
• 第1層鏈表不是一個單向鏈表，而是一個雙向鏈表。這是為了方便以倒序方式獲取一個范圍內的元素。
• 在skiplist中可以很方便地計算出每個元素的排名(rank)。

如何使用Redis進行限流

1、SortedSet
利用zset數(shù)據(jù)結構的score值來作為時間窗口，value保證唯一性即可,比如UUID或者雪花算法snowflake。

用一個zset結構記錄用戶的歷史行為，每一個行為都會作為zset中的一個key保存下來，同一個用戶的同一種行為用一個zset記錄。

為節(jié)省內存，只保留時間窗口內的行為記錄，如果用戶是冷用戶，窗口內的行為是空記錄，則這個zset可以從內存中移除。

通過統(tǒng)計窗口內的行為數(shù)量與閾值進行比較就可以得出當前行為是否允許。

代碼

public class SimpleRateLimiter {
    private Jedis jedis;

    public SimpleRateLimiter(Jedis jedis) {
        this.jedis = jedis;
    }

    // 指定用戶 user_id 的某個行為 action_key 在特定的時間內 period 只允許發(fā)生一定的次數(shù) max_co
    public boolean isActionAllowed(String userId, String actionKey, int period, int maxCount) {
        String key = String.format("hist:%s:%s", userId, actionKey);
        long nowTs = System.currentTimeMillis();
        Pipeline pipe = jedis.pipelined();
        pipe.multi();
        // value 沒有實際的意義，保證唯一就可以
        pipe.zadd(key, nowTs, "" + nowTs);
        pipe.zremrangeByScore(key, 0, nowTs - period * 1000);
        Response<Long> count = pipe.zcard(key);
        pipe.expire(key, period + 1);
        pipe.exec();
        pipe.close();
        return count.get() <= maxCount;
    }

    public static void main(String[] args) {
        Jedis jedis = new Jedis();
        SimpleRateLimiter limiter = new SimpleRateLimiter(jedis);
        for (int i = 0; i < 20; i++) {
           // 調用這個接口 , 一分鐘內只允許最多回復 5 個帖子
           System.out.println(limiter.isActionAllowed("laoqian", "reply", 60, 5));
        }
    }
}

缺點：因為它要記錄時間窗口內所有的行為記錄， 如果這個量很大，比如“限定 60s 內操作不得超過 100 萬次”之類，它是不適合做這樣的限流的，因為會消耗大量的存儲空間。

Redission版本

public static void main(String[] args) {
        RedissonClient redisson = Redisson.create();

        Config config = new Config();
        config.useSingleServer().setAddress("redis://127.0.0.1:6379");

        String id = "123";
        RBucket<Boolean> bucket = redisson.getBucket("blackId:" + id);
        // 是否是黑名單得
        if(bucket.get() == true){
            // 或者返回最近的一個請求的緩存
            return;
        }

        long nanoTime = System.nanoTime();
        RScoredSortedSet<Object> zset = redisson.getScoredSortedSet(id);
        zset.expire(10, TimeUnit.SECONDS);
        zset.add(nanoTime, nanoTime);
        zset.removeRangeByScore(0, true,
                nanoTime - 10 * 1000 * 1000 * 1000, true);
        int size = zset.size();
        // 超過了10次，則進入黑名單
        if(size > 10){
            // 加入黑名單，30秒之后不能再訪問
            bucket.set(true, 30, TimeUnit.SECONDS);
            // 或者返回最近的一個請求的緩存
            return;
        }
        // 放行
    }

不使用Redis的單機Java版本

public class TestLimitFlow {

   private Lock lock = new ReentrantLock();

   private Map<String, Entity> map = new ConcurrentHashMap<>();

   public TestLimitFlow() {
       new Thread(() -> {
           while (true) {
               System.out.println(map.size());
               map.forEach((key, value) -> {
                   long now = System.nanoTime();
                   if (value.expireTime < now) {
                       map.remove(key);
                   }
               });
               try {
                   TimeUnit.SECONDS.sleep(1);
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }
           }

       }).start();
   }


   class Entity<T> {
       public long expireTime;
       T o;
   }

   public void interceptor(String ip) throws InterruptedException {
       lock.lock();
       Entity e = map.get(ip);
       long nanoTime = System.nanoTime();
       if (e == null) {
           TreeSet<Long> treeSet = new TreeSet<>();
           e = new Entity<TreeSet<Long>>();
           e.expireTime = nanoTime + 10L * 1000 * 1000 * 1000;
           e.o = treeSet;
           map.put(ip, e);
       }
       lock.unlock();
       synchronized (e) {
           e.expireTime = nanoTime + 10L * 1000 * 1000 * 1000;
           TreeSet<Long> zset = (TreeSet<Long>) e.o;
           zset.add(nanoTime);
           List<Long> deleteKey = Lists.newArrayList();
           for(Long item : zset){
               if(item < nanoTime - 10L * 1000 * 1000 * 1000){
                   System.out.println("true");
//                    zset.remove(item);
                   deleteKey.add(item);
               }else {
                   break;
               }
           }
           deleteKey.forEach(item -> {
               zset.remove(item);
           });
           System.out.println(zset.size());
           if (zset.size() > 10) {
               // 加入黑名單
               System.out.println("黑名單：" + ip + ":" + zset.size());
           }
       }

   }

   public static void main(String[] args) throws InterruptedException {
       TestLimitFlow testXL = new TestLimitFlow();
       for(int j = 0; j < 10;j++){
           int finalJ = j;
           new Thread(){
               @Override
               public void run() {
                   for (int i = 0; i < 12; i++) {
//            TimeUnit.SECONDS.sleep(1);
                       try {
                           testXL.interceptor("localhost" + finalJ);
                       } catch (InterruptedException e) {
                           e.printStackTrace();
                       }
                   }
               }
           }.start();
       }


   }
}

2、令牌桶算法

如何使用 Redis 實現(xiàn)分布式鎖？

分布式鎖要解決的問題

• 互斥性
• 安全性
• 死鎖
• 容錯

版本1

使用 SEATNX key value:如果key不存在，則創(chuàng)建并賦值

@Service
public class RedisSeckillServiceImpl {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    public String deductStock() {
        String lockKey = "lambo";
        try {
            String value = "value";
            Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, value);
            if (!result) {
                return "error_code";
            }
            // 此處機器宕機，將發(fā)生死鎖
            int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
            if (stock > 0) {
                int realStock = stock - 1;
                stringRedisTemplate.opsForValue().set("stock", realStock + "");
                System.out.println("扣減成功，剩余庫存：" + realStock);
            } else {
                System.out.println("扣減失敗，庫存不足");
            }
        } catch (Exception e) {

        } finally {
            stringRedisTemplate.delete(lockKey);
        }
        return "end";
    }

}

分析

在setnx后如果機器宕機，鎖得不到釋放，就會產生死鎖

版本2

給key設置過期時間

EXPIRE key seconds

@Service
public class RedisSeckillServiceImpl {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    /**
     * 設置過期時間，防止宕機帶來的死鎖
     *
     * @return
     */
    public String deductStock2() {
        String lockKey = "lambo";
        try {
            String value = "value";
            Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, value);
            // 設置過期時間，但是set和expire不是原子性操作，還沒有設置過期時間，機器宕機了
            stringRedisTemplate.expire(lockKey, 10, TimeUnit.SECONDS);
            if (!result) {
                return "error_code";
            }
        
            int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
            if (stock > 0) {
                int realStock = stock - 1;
                stringRedisTemplate.opsForValue().set("stock", realStock + "");
                System.out.println("扣減成功，剩余庫存：" + realStock);
            } else {
                System.out.println("扣減失敗，庫存不足");
            }
        } catch (Exception e) {

        } finally {
            stringRedisTemplate.delete(lockKey);
        }
        return "end";
    }
   
}

缺點

原子性等不到滿足，死鎖問題依舊會出現(xiàn)

版本3

使用原子性操作
SETNX key value [EX seconds] [PX milliseconds] [NX|XX]
NX:只在鍵不存在時，才對鍵進行設置
XX:只在鍵存在時，才對鍵進行設置

@Service
public class RedisSeckillServiceImpl {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;
  
    /**
     * set+expire變更成原子操作
     *
     * @return
     */
    public String deductStock3() {
        String lockKey = "lambo";

        try {
            String value = "value";
            // 原子操作
            Boolean result = stringRedisTemplate.opsForValue().setIfPresent(lockKey, value, 10, TimeUnit.SECONDS);
            if (!result) {
                return "error_code";
            }
            
            int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
            if (stock > 0) {
                int realStock = stock - 1;
                stringRedisTemplate.opsForValue().set("stock", realStock + "");
                System.out.println("扣減成功，剩余庫存：" + realStock);
            } else {
                System.out.println("扣減失敗，庫存不足");
            }
        } catch (Exception e) {

        } finally {
            
            // 線程可能會釋放其他線程的鎖，無法保證安全性
            stringRedisTemplate.delete(lockKey);
        }
        return "end";
    }
}

分析

比如線程1設置key10秒過期，執(zhí)行業(yè)務需要15s，10秒后線程2進入，由于key過期自動刪除后，線程2拿到鎖，執(zhí)行業(yè)務需要8秒，在第15秒的時候線程一將鎖釋放了，但是此時線程也還沒操作完，這樣其他線程又可以搶到鎖了

版本4

給線程設置唯一值，釋放鎖的時候只有加鎖的線程才可以釋放鎖

@Service
public class RedisSeckillServiceImpl {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;


    /**
     * 線程只能釋放自己加的鎖，不能釋放別的線程加的鎖
     *
     * @return
     */
    public String deductStock4() {
        String lockKey = "lambo";
        String clientId = UUID.randomUUID().toString();
        try {
            // 原子操作
            Boolean result =
            // 設置唯一值
            stringRedisTemplate.opsForValue().setIfPresent(lockKey, clientId, 10, TimeUnit.SECONDS);
            if (!result) {
                return "error_code";
            }
          
            int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
            if (stock > 0) {
                int realStock = stock - 1;
                stringRedisTemplate.opsForValue().set("stock", realStock + "");
                System.out.println("扣減成功，剩余庫存：" + realStock);
            } else {
                System.out.println("扣減失敗，庫存不足");
            }
        } catch (Exception e) {

        } finally {
            if (clientId.equals(stringRedisTemplate.opsForValue().get(lockKey))) {
                stringRedisTemplate.delete(lockKey);
            }
        }
        return "end";
    }

}

分析

此時線程是只能釋放自己加的鎖了，無法釋放其他線程的鎖，但是有可能業(yè)務本身執(zhí)行所需要的時間就是超過過期時間，而且這個時間是無法預估的，此時我們一種機制，每隔一段時間去判斷線程是否還持有鎖，如果持有鎖則延長鎖的過期時間。

版本5

使用redisson的看門狗機制

@Service
public class RedisSeckillServiceImpl {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private RedissonClient redissonClient;

    /**
     * 基于redisson設置分布式鎖
     *
     * @return
     */
    public String deductStock5() {
        String lockKey = "lambo";
        RLock redissonLock = redissonClient.getLock(lockKey);
        try {
            redissonLock.lock();
            int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
            if (stock > 0) {
                int realStock = stock - 1;
                stringRedisTemplate.opsForValue().set("stock", realStock + "");
                System.out.println("扣減成功，剩余庫存：" + realStock);
            } else {
                System.out.println("扣減失敗，庫存不足");
            }
        } catch (Exception e) {

        } finally {
            redissonLock.unlock();
        }
        return "end";

    }
}

如果Redis有1億個key，使用keys命令是否會影響線上服務

首先keys pattern命令可以從redis中查找所有符合給定模式pattern的key
如果Redis有1億個key，使用keys命令是會影響線上服務的，因為Redis是單線程模型，當然可以部署多個節(jié)點。
但是redis提供從海量數(shù)據(jù)里查詢某一固定前綴的key的命令：
SCAN cursor [MATCH pattern] [COUNT count]
1、基于游標的迭代器。需要基于上一次的游標延續(xù)之前的迭代過程
2、以0作為游標開始一次新的迭代，直到命令返回游標0完成一次遍歷
3、不保證每次執(zhí)行都返回某個給定數(shù)量的元素，支持模糊查詢
4、一次返回的數(shù)量不可控，只能是大概率符合count數(shù)
例如 scan 0 match k* count 1;

如何使用 Redis 實現(xiàn)消息隊列？

一般使用 list 結構作為隊列，rpush 生產消息，lpop 消費消息。當 lpop 沒有消息的時候，要適當 sleep 一會再重試。

可不可以不用 sleep 呢？

list 還有個指令叫 blpop ，在沒有消息的時候，它會阻塞住直到消息到來。

能不能生產一次消費多次呢？

使用 pub / sub 主題訂閱者模式，發(fā)送者(pub)發(fā)送消息，訂閱者(sub)接收消息?？梢詫崿F(xiàn) 1:N 的消息隊列。

• 訂閱命令
  subscribe channelName
• 發(fā)布命令
  publish channelName value

pub / sub 有什么缺點？

在消費者下線的情況下，生產的消息會丟失，得使用專業(yè)的消息隊列如 rabbitmq 等。

如何實現(xiàn)延時隊列？

使用 sortedset ，拿時間戳作為 score ，消息內容作為 key 調用 zadd 來生產消息，消費者用 zrangebyscore 指令獲取 N 秒之前的數(shù)據(jù)輪詢進行處理。

談談緩存擊穿，緩存雪崩，緩存穿透

緩存擊穿(單個key失效同時有大量請求)

定義

緩存中的key一般有設置過期時間，如果某個key過期了，在這個時候，有大量的并發(fā)請求訪問這個key，則這寫請求都會進入到DB中，導致DB瞬間壓力過大，壓垮DB

解決方案

1、設置互斥鎖，mutex。當緩存失效的時候不是立即去訪問數(shù)據(jù)庫，而是使用分布式鎖，比如redis，redission,zookeeper,

缺點

可能造成死鎖，或者線程池阻塞

緩存穿透（key本身不存在于數(shù)據(jù)庫）

定義

數(shù)據(jù)庫中不存在的key，自然而然緩存中而不存在，如果有大量地訪問不存在的key的請求，請求會直接到數(shù)據(jù)庫，壓垮數(shù)據(jù)庫

解決方案

1、布隆過濾器
布隆過濾器是一個非常神奇的數(shù)據(jù)結構，通過它我們可以非常方便地判斷一個給定數(shù)據(jù)是否存在與海量數(shù)據(jù)中。我們需要的就是判斷 key 是否合法，有沒有感覺布隆過濾器就是我們想要找的那個“人”。具體是這樣做的：把所有可能存在的請求的值都存放在布隆過濾器中，當用戶請求過來，我會先判斷用戶發(fā)來的請求的值是否存在于布隆過濾器中。不存在的話，直接返回請求參數(shù)錯誤信息給客戶端，存在的話才會走下面的流程。
具體可看這篇文章
https://github.com/Snailclimb/JavaGuide/blob/master/docs/dataStructures-algorithms/data-structure/bloom-filter.md

2、不存在的key，在緩存中的value值設置為null，并且設置過期時間，

java偽代碼如下

public Object getObjectInclNullById(Integer id) {
    // 從緩存中獲取數(shù)據(jù)
    Object cacheValue = cache.get(id);
    // 緩存為空
    if (cacheValue != null) {
        // 從數(shù)據(jù)庫中獲取
        Object storageValue = storage.get(key);
        // 緩存空對象
        cache.set(key, storageValue);
        // 如果存儲數(shù)據(jù)為空，需要設置一個過期時間(300秒)
        if (storageValue == null) {
            // 必須設置過期時間，否則有被攻擊的風險
            cache.expire(key, 60 * 5);
        }
        return storageValue;
    }
    return cacheValue;

緩存雪崩(大量key同時失效)

定義

緩存中大量的key在同一時間失效，這時候大量的請求就會直接到數(shù)據(jù)庫，使數(shù)據(jù)庫的壓力過大

解決方案

1、在過期時間上添加隨機值，把緩存失效的時間錯開，這樣的話失效的時間的重復率就降低了，降低了集體失效的概率

Redis 有幾種數(shù)據(jù)“過期”策略？
Redis key過期的方式有三種：

首先刪除策略有三種

定時刪除

在設置鍵的過期時間的同時，創(chuàng)建一個定時器(timer)，讓定時器在鍵的過期時間來臨，立即執(zhí)行對鍵的刪除操作

優(yōu)點：對內存友好

缺點：對CPU不友好

惰性刪除

放任鍵過期時間不管，但是每次從鍵空間中獲取鍵時，都檢查取得的鍵是否過期，如果過期的話就刪除該鍵；如果沒有過期就返回該鍵

優(yōu)點：對CPU友好

缺點：對內存不友好，造成內存泄漏

定期刪除

每隔一段時間，程序就對數(shù)據(jù)庫進行一次檢查沒刪除里面的過期鍵。至于要刪除多少過期鍵，以及檢查多少個數(shù)據(jù)庫，則由算法決定

優(yōu)點：是對定時刪除和惰性刪除的一種整合和折中

缺點：難以缺點刪除操作執(zhí)行的時長和頻率。執(zhí)行太頻繁會退化成定時策略，執(zhí)行太少又會退化成惰性刪除策略

定時刪除和定期刪除為主動刪除策略，惰性刪除為被動刪除策略

Redis刪除策略

被動刪除：當讀/寫一個已經過期的key時，會觸發(fā)惰性刪除策略，直接刪除掉這個過期key

主動刪除（定期刪除策略）：由于惰性刪除策略無法保證冷數(shù)據(jù)被及時刪掉，所以Redis會定期主動淘汰一批已過期的key

當前已用內存超過maxmemory限定時，觸發(fā)主動清理策略(即下面的淘汰策略)

Redis 有哪幾種數(shù)據(jù)“淘汰”策略？

==Redis 提供了 6 種數(shù)據(jù)淘汰策略：==

• volatile-lru 只對設置了過期時間的key進行LRU
• volatile-ttl 刪除即將過期了的key
• volatile-random
  只對設置了過期時間的key進行隨機刪除
• allkeys-lru
  對所有key進行LRU算法刪除
• allkeys-random
  對所有key進行隨機刪除
• no-enviction
  永不過期，直接拋錯

Redis持久化機制有哪些？

Redis有兩種持久化機制，AOF和RDB。

• AOF，記錄每次寫請求的命令，以追加的方式在文件尾部追加，直接在尾部追加，效率比較高。
  對于操作系統(tǒng)來說，不是每次寫都直接寫到磁盤，操作系統(tǒng)自己會有一層cache，redis寫磁盤的數(shù)據(jù)會先緩存在os cache里，redis每隔1秒調用一次操作系統(tǒng)的fsync操作，強制將os cache中的數(shù)據(jù)刷入AOF文件中。

當redis重啟的時候，就把AOF中記錄的命令重新執(zhí)行一遍就可以了，但是如果文件很大的話，執(zhí)行會耗費較多的時間，對于數(shù)據(jù)恢復來說耗時會多一點。

• RDB，是快照文件，每隔一定時間將redis內存中的數(shù)據(jù)生成一份完整的RDB快照文件，當redis重啟的時候直接加載數(shù)據(jù)即可，同樣的數(shù)據(jù)比AOF恢復的要快。

說說這兩種持久化機制各自的特點、優(yōu)缺點吧

==RDB優(yōu)點==

第一點就是他會生成多個數(shù)據(jù)文件，每個數(shù)據(jù)文件都代表了某一時刻redis中的全量數(shù)據(jù)，非常適合做冷備。
第二點，RDB持久化機制對redis對外提供的讀寫服務影響非常小，可以讓redis保持高性能，因為redis主進程只需要fork一個子進程，讓子進程執(zhí)行磁盤IO操作來進行RDB持久化即可。
第三點，相對于AOF持久化機制來說，直接基于RDB數(shù)據(jù)文件來重啟和恢復redis進程，更加快速。

RDB，就是一份數(shù)據(jù)文件，恢復的時候，直接加載到內存中即可。

==RBD的缺點==

1）故障時可能數(shù)據(jù)丟失的比AOF要多。
可能會因為Redis掛起而且是從當前值最近一次快照期間的數(shù)據(jù)

這個問題，也是rdb最大的缺點，就是不適合做第一優(yōu)先的恢復方案，如果你依賴RDB做第一優(yōu)先恢復方案，會導致數(shù)據(jù)丟失的比較多

2）RDB每次在fork子進程來執(zhí)行RDB快照數(shù)據(jù)文件生成的時候，如果數(shù)據(jù)文件特別大，可能會導致對客戶端提供的服務暫停數(shù)毫秒，或者甚至數(shù)秒。RDB是對內存數(shù)據(jù)的全量同步，數(shù)據(jù)量大會由于I/O而影響性能

所以一般不要讓RDB的間隔太長，否則每次生成的RDB文件太大了，對redis本身的性能可能會有影響的。

==AOF的優(yōu)點==

AOF持久化，記錄下除了查詢以外的所有變更數(shù)據(jù)量的指令，以append的形式追加保存到AOF文件中

1）AOF可以更好的保護數(shù)據(jù)不丟失
一般AOF會每隔1秒，通過一個后臺線程執(zhí)行一次fsync操作，最多丟失1秒鐘的數(shù)據(jù)。

每隔1秒，就執(zhí)行一次fsync操作，保證os cache中的數(shù)據(jù)寫入磁盤中。
redis進程掛了，最多丟掉1秒鐘的數(shù)據(jù).

2）AOF持久化性能高
AOF日志文件以append-only模式寫入，所以沒有任何磁盤尋址的開銷，寫入性能非常高，而且文件不容易破損，即使文件尾部破損，也很容易修復。

3）AOF日志文件即使過大的時候，出現(xiàn)后臺重寫操作，也不會影響客戶端的讀寫。
因為在rewrite log的時候，會對其中的指令進行壓縮，創(chuàng)建出一份需要恢復數(shù)據(jù)的最小日志出來。再創(chuàng)建新日志文件的時候，老的日志文件還是照常寫入。當新的merge后的日志文件ready的時候，再交換新老日志文件即可。

4）AOF日志文件的命令通過非常可讀的方式進行記錄，這個特性非常適合做災難性的誤刪除的緊急恢復。

比如某人不小心用flushall命令清空了所有數(shù)據(jù)，只要這個時候后臺rewrite還沒有發(fā)生，那么就可以立即拷貝AOF文件，將最后一條flushall命令給刪了，然后再將該AOF文件放回去，就可以通過恢復機制，自動恢復所有數(shù)據(jù)。

==AOF的缺點==

（1）對于同一份數(shù)據(jù)來說，AOF日志文件通常比RDB數(shù)據(jù)快照文件更大

（2）AOF開啟后，支持的寫QPS會比RDB支持的寫QPS低，因為AOF一般會配置成每秒fsync一次日志文件，當然，每秒一次fsync，性能也還是很高的。

如果你要保證一條數(shù)據(jù)都不丟，也是可以的，AOF的fsync設置成沒寫入一條數(shù)據(jù)，fsync一次，但是那樣導致redis的QPS大幅度下降。

（3）以前AOF發(fā)生過bug，就是通過AOF記錄的日志，進行數(shù)據(jù)恢復的時候，沒有恢復一模一樣的數(shù)據(jù)出來。

所以說，類似AOF這種較為復雜的基于命令日志/merge/回放的方式，比基于RDB每次持久化一份完整的數(shù)據(jù)快照文件的方式，更加脆弱一些，容易有bug。

不過AOF就是為了避免rewrite過程導致的bug，因此每次rewrite并不是基于舊的指令日志進行merge的，而是基于當時內存中的數(shù)據(jù)進行指令的重新構建，這樣健壯性會好很多。

（4）唯一的比較大的缺點，其實就是做數(shù)據(jù)恢復的時候，會比較慢，做冷備不太合適。

AOF持久化過程

• 調用fork（）創(chuàng)建一個子進程

• 子進程將新的AOF寫到一個臨時文件里，不依賴原來的AOF文件

• 主進程持續(xù)將新的變動寫到內存和原來的AOF文件里

• 主進程獲取子進程重新AOF的完成信息，往新AOF增量變動

• 使用新的AOF文件替換掉舊的AOF文件

==主從復制==

首先說一下Redis Sentinel是怎么工作的？重點描述一下故障轉移的過程

• Sentinel會以每秒一次的頻率向所有與它創(chuàng)建了命令連接的實例（包括主服務器，從服務器，其他Sentinel在內）發(fā)送PING命令

• 如果一個實例在down-after-milliseconds毫秒內沒有向Sentinel返回有效回復，則該Sentinel任務該實例處于下線狀態(tài)，稱為主觀下線

• 當Sentinel從其他Sentinel那里收到足夠數(shù)量的已下線的判斷后，Sentinel就會將從服務器從主觀下線判定為客觀下線，并對主服務器進行故障轉移操作

• 當一個主服務器被判斷為客觀下線后，監(jiān)視這個下線主服務器的各個Sentinel會進行協(xié)商，選舉出一個領頭的Sentinel，并由它對下線主服務器進行故障轉移操作。

故障轉移操作主要步驟：

• 在已下線主服務器下的所有從服務器器里，挑選出一個從服務器，并將其轉換成新的主服務器

• 讓已下線主服務器下的其他從服務器改為復制新的主服務器

• 將已下線主服務器設置為新的主服務器的從服務器，當這個舊的主服務器重新上線時，它就會成為新的主服務器的從服務器

故障轉移時會從剩下的slave選舉一個新的master，被選舉為master的標準是什么？

• Sentinel會將主服務器和從服務器的信息保存在一個列表中

• 主服務器下線時，會刪除列表中所有處于下線或者斷線狀態(tài)的從服務器

• 刪除列表中所有最近5秒內沒有回復過領頭Sentinel的INFO命令的從服務器，

• 刪除所有與已下線的主服務器連接斷開炒貨down-after-milliseconds毫秒的從服務器，

• 然后根據(jù)優(yōu)先級（從高到低），復制偏移量（從高到低），運行id（從小到大）進行排序選出領頭的Sentinel

執(zhí)行切換的那個哨兵在完成故障轉移后會做什么？

會進行configuraiton配置信息傳播。

哨兵完成切換之后，會在自己本地更新生成最新的master配置，然后通過pub/sub消息機制同步給其他的哨兵。

==Redis集群==