Redis實戰篇

• 2019 年 10 月 23 日
• 筆記

Redis實戰篇

1 Redis 客戶端

1.1 客戶端通訊 原理

客戶端和伺服器通過 TCP 連接來進行數據交互， 伺服器默認的埠號為 6379 。
客戶端和伺服器發送的命令或數據一律以 rn （CRLF 回車+換行）結尾。

如果使用 wireshark 對 jedis 抓包：
環境：Jedis 連接到虛擬機 202，運行 main，對 VMnet8 抓包。
過濾條件：ip.dst==192.168.8.202 and tcp.port in {6379}
set qingshan 抓包：

可以看到實際發出的數據包是：

*3rn$3rnSETrn$8rnqingshanrn$4rn2673rn

get qingshan 抓包：

*2rn$3rnGETrn$8rnqingshanrn

​ 客戶端跟 Redis 之間 使用一種特殊的編碼格式（在 AOF 文件裡面我們看到了），叫做 Redis Serialization Protocol (Redis 序列化協議)。特點：容易實現、解析快、可讀性強。客戶端發給服務端的消息需要經過編碼，服務端收到之後會按約定進行解碼，反之亦然。

基於此，我們可以自己實現一個 Redis 客戶端。
參考：myclient.MyClient.java
1、建立 Socket 連接
2、OutputStream 寫入數據（發送到服務端）
3、InputStream 讀取數據（從服務端介面）
基於這種協議，我們可以用 Java 實現所有的 Redis 操作命令。當然，我們不需要這麼做，因為已經有很多比較成熟的 Java 客戶端，實現了完整的功能和高級特性，並且提供了良好的性能。

https://redis.io/clients#java
官網推薦的 Java 客戶端有 3 個 Jedis，Redisson 和 Luttuce。

Spring 連接 Redis 用的是什麼？RedisConnectionFactory 介面支援多種實現，例如 ： JedisConnectionFactory 、 JredisConnectionFactory 、LettuceConnectionFactory、SrpConnectionFactory。

1.2 Jedis

https://github.com/xetorthio/jedis

1.2.1 特點

Jedis 是我們最熟悉和最常用的客戶端。輕量，簡潔，便於集成和改造。

public static void main(String[] args) {  Jedis jedis = new Jedis("127.0.0.1", 6379);  jedis.set("qingshan", "2673");  System.out.println(jedis.get("qingshan"));  jedis.close();  }

Jedis 多個執行緒使用一個連接的時候執行緒不安全。可以使用連接池，為每個請求創建不同的連接，基於 Apache common pool 實現。跟資料庫一樣，可以設置最大連接數等參數。Jedis 中有多種連接池的子類。

例如：

public class ShardingTest {      public static void main(String[] args) {          JedisPoolConfig poolConfig = new JedisPoolConfig();            // Redis伺服器          JedisShardInfo shardInfo1 = new JedisShardInfo("127.0.0.1", 6379);          JedisShardInfo shardInfo2 = new JedisShardInfo("192.168.8.205", 6379);            // 連接池          List<JedisShardInfo> infoList = Arrays.asList(shardInfo1, shardInfo2);          ShardedJedisPool jedisPool = new ShardedJedisPool(poolConfig, infoList);            ShardedJedis jedis = null;          try{              jedis = jedisPool.getResource();              for(int i=0; i<100; i++){                  jedis.set("k"+i, ""+i);              }              for(int i=0; i<100; i++){                  Client client = jedis.getShard("k"+i).getClient();                  System.out.println("取到值："+jedis.get("k"+i)+"，"+"當前key位於：" + client.getHost() + ":" + client.getPort());              }            }finally{              if(jedis!=null) {                  jedis.close();              }          }      }  }

​ Jedis 有 4 種工作模式：單節點、分片、哨兵、集群。
​ 3 種請求模式：Client、Pipeline、事務。Client 模式就是客戶端發送一個命令，阻塞等待服務端執行，然後讀取 返回結果。Pipeline 模式是一次性發送多個命令，最後一次取回所有的返回結果，這種模式通過減少網路的往返時間和 io 讀寫次數，大幅度提高通訊性能。第三種是事務模式。Transaction 模式即開啟 Redis 的事務管理，事務模式開啟後，所有的命令（除了 exec，discard，multi 和 watch）到達服務端以後不會立即執行，會進入一個等待隊列。

1.2.2 Sentinel 獲取 連接原理

​ 問題：Jedis 連接 Sentinel 的時候，我們配置的是全部哨兵的地址。Sentinel 是如何返回可用的 master 地址的呢？

在構造方法中：

pool = new JedisSentinelPool(masterName, sentinels);

調用了：

HostAndPort master = initSentinels(sentinels, masterName);

查看：

private HostAndPort initSentinels(Set<String> sentinels, final String masterName) {  ​      HostAndPort master = null;      boolean sentinelAvailable = false;      ​      log.info("Trying to find master from available Sentinels...");      // 有多個 sentinels,遍歷這些個 sentinels      for (String sentinel : sentinels) {      // host:port 表示的 sentinel 地址轉化為一個 HostAndPort 對象。      final HostAndPort hap = HostAndPort.parseString(sentinel);      ​      log.fine("Connecting to Sentinel " + hap);      ​      Jedis jedis = null;      try {      // 連接到 sentinel          jedis = new Jedis(hap.getHost(), hap.getPort());          // 根據 masterName 得到 master 的地址，返回一個 list，host= list[0], port =// list[1]          List<String> masterAddr = jedis.sentinelGetMasterAddrByName(masterName);      ​          // connected to sentinel...          sentinelAvailable = true;      ​          if (masterAddr == null || masterAddr.size() != 2) {              log.warning("Can not get master addr, master name: " + masterName + ". Sentinel: " + hap + ".");              continue;          }          // 如果在任何一個 sentinel 中找到了 master，不再遍歷 sentinels          master = toHostAndPort(masterAddr);          log.fine("Found Redis master at " + master);          break;      } catch (JedisException e) {          // resolves #1036, it should handle JedisException there's another chance          // of raising JedisDataException          log.warning("Cannot get master address from sentinel running @ " + hap + ". Reason: " + e          + ". Trying next one.");      } finally {          if (jedis != null) {              jedis.close();          }      }  }          // 到這裡，如果 master 為 null，則說明有兩種情況，一種是所有的 sentinels 節點都 down 掉了，一種是 master節點沒有被存活的 sentinels 監控到  if (master == null) {      if (sentinelAvailable) {      // can connect to sentinel, but master name seems to not      // monitored      throw new JedisException("Can connect to sentinel, but " + masterName + " seems to be not monitored...");      } else {          throw new JedisConnectionException("All sentinels down, cannot determine where is " + masterName + " master is running...");      }  }  // 如果走到這裡，說明找到了 master 的地址  log.info("Redis master running at " + master + ", starting Sentinel listeners...");  // 啟動對每個 sentinels 的監聽為每個 sentinel 都啟動了一個監聽者 MasterListener。MasterListener 本身是一個執行緒，它會去訂閱 sentinel 上關於 master 節點地址改變的消息。  for (String sentinel : sentinels) {      final HostAndPort hap = HostAndPort.parseString(sentinel);      MasterListener masterListener = new MasterListener(masterName, hap.getHost(), hap.getPort());      // whether MasterListener threads are alive or not, process can be stopped      masterListener.setDaemon(true);      masterListeners.add(masterListener);      masterListener.start();  }  ​  return master;  }

1.2.3 Cluster 獲取 連接原理

​ 問題：使用 Jedis 連接 Cluster 的時候，我們只需要連接到任意一個或者多個 redisgroup 中的實例地址，那我們是怎麼獲取到需要操作的 Redis Master 實例的？
​ 關鍵問題：在於如何存儲 slot 和 Redis 連接池的關係。
​ 1、程式啟動初始化集群環境，讀取配置文件中的節點配置，無論是主從，無論多少個，只拿第一個，獲取 redis 連接實例（後面有個 break）。

// redis.clients.jedis.JedisClusterConnectionHandler#initializeSlotsCache  private void initializeSlotsCache(Set<HostAndPort> startNodes, GenericObjectPoolConfig poolConfig, String password)  {      for (HostAndPort hostAndPort : startNodes) {          // 獲取一個 Jedis 實例          Jedis jedis = new Jedis(hostAndPort.getHost(), hostAndPort.getPort());          if (password != null) {              jedis.auth(password);          }          try {              // 獲取 Redis 節點和 Slot 虛擬槽              cache.discoverClusterNodesAndSlots(jedis);              // 直接跳出循環              break;          } catch (JedisConnectionException e) {          // try next nodes          } finally {          if (jedis != null) {              jedis.close();          }      }  }

2、用獲取的 redis 連接實例執行 clusterSlots ()方法，實際執行 redis 服務端 clusterslots 命令，獲取虛擬槽資訊。
該集合的基本資訊為[long, long, List, List], 第一，二個元素是該節點負責槽點的起始位置，第三個元素是主節點資訊，第四個元素為主節點對應的從節點資訊。該 list 的基本資訊為[string,int,string],第一個為 host 資訊，第二個為 port 資訊，第三個為唯一id。

​ 3、獲取有關節點的槽點資訊後，調用 getAssignedSlotArray(slotinfo)來獲取所有的槽點值。
​ 4、再獲取主節點的地址資訊，調用 generateHostAndPort(hostInfo)方法，生成一個 ostAndPort 對象。
​ 5、再根據節點地址資訊來設置節點對應的 JedisPool，即設置 Map<String,JedisPool> nodes 的值。
​ 接下來判斷若此時節點資訊為主節點資訊時，則調用 assignSlotsToNodes 方法，設置每個槽點值對應的連接池，即設置 Map<Integer, JedisPool> slots 的值。

public void discoverClusterNodesAndSlots(Jedis jedis) {      w.lock();  ​      try {          reset();          // 獲取節點集合          List<Object> slots = jedis.clusterSlots();          // 遍歷 3 個 master 節點          for (Object slotInfoObj : slots) {          // slotInfo 槽開始，槽結束，主，從          // {[0,5460,7291,7294],[5461,10922,7292,7295],[10923,16383,7293,7296]}          List<Object> slotInfo = (List<Object>) slotInfoObj;          // 如果<=2，代表沒有分配 slot          if (slotInfo.size() <= MASTER_NODE_INDEX) {              continue;          }          // 獲取分配到當前 master 節點的數據槽，例如 7291 節點的{0,1,2,3……5460}          List<Integer> slotNums = getAssignedSlotArray(slotInfo);          // hostInfos          int size = slotInfo.size(); // size 是 4，槽最小最大，主，從          // 第 3 位和第 4 位是主從埠的資訊          for (int i = MASTER_NODE_INDEX; i < size; i++) {              List<Object> hostInfos = (List<Object>) slotInfo.get(i);              if (hostInfos.size() <= 0) {                  continue;              }              // 根據 IP 埠生成 HostAndPort 實例              HostAndPort targetNode = generateHostAndPort(hostInfos);              // 據HostAndPort解析出ip:port的key值，再根據key從快取中查詢對應的jedisPool實例。如果沒有jedisPool實例，就創建 JedisPool 實例，最後放入快取中。nodeKey 和 nodePool 的關係              setupNodeIfNotExist(targetNode);              // 把 slot 和 jedisPool 快取起來（16384 個），key 是 slot 下標，value 是連接池              if (i == MASTER_NODE_INDEX) {                  assignSlotsToNode(slotNums, targetNode);              }          }        } finally {          w.unlock();      }  }  

從集群環境存取值：
1、把 key 作為參數，執行 CRC16 演算法，獲取 key 對應的 slot 值。
2、通過該 slot 值，去 slots 的 map 集合中獲取 jedisPool 實例。
3、通過 jedisPool 實例獲取 jedis 實例，最終完成 redis 數據存取工作。

1.2.4 pipeline

​ 我們看到 set 2 萬個 key 用了好幾分鐘，這個速度太慢了，完全沒有把 Redis 10萬的 QPS 利用起來。但是單個命令的執行到底慢在哪裡？

1.2.4.1 慢在 哪裡？

​ Redis 使用的是客戶端/伺服器（C/S）模型和請求/響應協議的 TCP 伺服器。這意味著通常情況下一個請求會遵循以下步驟：

• 客戶端向服務端發送一個查詢請求，並監聽 Socket 返回，通常是以阻塞模式，等待服務端響應。

• 服務端處理命令，並將結果返回給客戶端。

​ Redis 客戶端與 Redis 伺服器之間使用 TCP 協議進行連接，一個客戶端可以通過一個 socket 連接發起多個請求命令。每個請求命令發出後 client 通常會阻塞並等待 redis伺服器處理，redis 處理完請求命令後會將結果通過響應報文返回給 client，因此當執行多條命令的時候都需要等待上一條命令執行完畢才能執行。執行過程如圖：

​ Redis 本身提供了一些批量操作命令，比如 mget，mset，可以減少通訊的時間，但是大部分命令是不支援 multi 操作的，例如 hash 就沒有.

​ 由於通訊會有網路延遲，假如 client 和 server 之間的包傳輸時間需要 10 毫秒，一次交互就是 20 毫秒（RTT：Round Trip Time）。這樣的話，client 1 秒鐘也只能也只能發送 50 個命令。這顯然沒有充分利用 Redis 的處理能力。另外一個，Redis 服務端執行 I/O 的次數過多。

1.2.4.2 Pipeline 管道

https://redis.io/topics/pipelining
那我們能不能像資料庫的 batch 操作一樣，把一組命令組裝在一起發送給 Redis 服務端執行，然後一次性獲得返回結果呢？這個就是 Pipeline 的作用。Pipeline 通過一個隊列把所有的命令快取起來，然後把多個命令在一次連接中發送給伺服器。

先來看一下效果（先 flushall）：
PipelineSet.java，PipelineGet.java
要實現 Pipeline，既要服務端的支援，也要客戶端的支援。對於服務端來說，需要能夠處理客戶端通過一個 TCP 連接發來的多個命令，並且逐個地執行命令一起返回 。

​ 對於客戶端來說，要把多個命令快取起來，達到一定的條件就發送出去，最後才處理 Redis 的應答（這裡也要注意對客戶端記憶體的消耗）。

​ jedis-pipeline 的 client-buffer 限制：8192bytes，客戶端堆積的命令超過 8192bytes 時，會發送給服務端。

源碼：redis.clients.util.RedisOutputStream.java

public RedisOutputStream(final OutputStream out) {      this(out, 8192);  }

​ pipeline 對於命令條數沒有限制，但是命令可能會受限於 TCP 包大小。
​ 如果 Jedis 發送了一組命令，而發送請求還沒有結束，Redis 響應的結果會放在接緩衝區。如果接收緩衝區滿了，jedis 會通知 redis win=0，此時 redis 不會再發送結果給 jedis 端，轉而把響應結果保存在 Redis 服務端的輸出緩衝區中。

​ 輸出緩衝區的配置：redis.conf

​ client-output-buffer-limit

client-output-buffer-limit normal 0 0 0  client-output-buffer-limit replica 256mb 64mb 60  client-output-buffer-limit pubsub 32mb 8mb 60

每個客戶端使用的輸出緩衝區的大小可以用 client list 命令查看

redis> client list

id=5 addr=192.168.8.1:10859 fd=8 name= age=5 idle=0 flags=N db=0 sub=0 psub=0 multi=-1 qbuf=5 qbuf-free=32763  obl=16380 oll=227 omem=4654408 events=rw cmd=set

• obl ： 輸出緩衝區的長度（位元組為單位， 0 表示沒有分配輸出緩衝區）
• oll ： 輸出列表包含的對象數量（當輸出緩衝區沒有剩餘空間時，命令回復會以字元串對象的形式被入隊到這個隊列里）
• omem ： 輸出緩衝區和輸出列表佔用的記憶體總量

1.2.4.3 使用 場景

​ Pipeline 適用於什麼場景呢？
​ 如果某些操作需要馬上得到 Redis 操作是否成功的結果，這種場景就不適合。
​ 有些場景，例如批量寫入數據，對於結果的實時性和成功性要求不高，就可以用Pipeline

1.2.5 Jedis 實現分散式 鎖

原文地址：https://redis.io/topics/distlock
中文地址：http://redis.cn/topics/distlock.html

分散式鎖的基本特性或者要求：
1、互斥性：只有一個客戶端能夠持有鎖。
2、不會產生死鎖：即使持有鎖的客戶端崩潰，也能保證後續其他客戶端可以獲取鎖。
3、只有持有這把鎖的客戶端才能解鎖。

distlock.DistLock.java

/**       * 嘗試獲取分散式鎖       * @param jedis Redis客戶端       * @param lockKey 鎖       * @param requestId 請求標識       * @param expireTime 超期時間       * @return 是否獲取成功       */      public static boolean tryGetDistributedLock(Jedis jedis, String lockKey, String requestId, int expireTime) {          // set支援多個參數 NX（not exist） XX（exist） EX（seconds） PX（million seconds）          String result = jedis.set(lockKey, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);          if (LOCK_SUCCESS.equals(result)) {              return true;          }          return false;      }

參數解讀：
1、lockKey 是 Redis key 的名稱，也就是誰添加成功這個 key 代表誰獲取鎖成功。
2、requestId 是客戶端的 ID（設置成 value），如果我們要保證只有加鎖的客戶端才能釋放鎖，就必須獲得客戶端的 ID（保證第 3 點）。
3、SET_IF_NOT_EXIST 是我們的命令裡面加上 NX（保證第 1 點）。
4、SET_WITH_EXPIRE_TIME，PX 代表以毫秒為單位設置 key 的過期時間（保證第 2 點）。expireTime 是自動釋放鎖的時間，比如 5000 代表 5 秒。

釋放鎖，直接刪除 key 來釋放鎖可以嗎？就像這樣：

public static void wrongReleaseLock1(Jedis jedis, String lockKey) {      jedis.del(lockKey);  }

沒有對客戶端 requestId 進行判斷，可能會釋放其他客戶端持有的鎖。
先判斷後刪除呢？

public static void wrongReleaseLock2(Jedis jedis, String lockKey, String requestId) {      // 判斷加鎖與解鎖是不是同一個客戶端      if (requestId.equals(jedis.get(lockKey))) {          // 若在此時，這把鎖突然不是這個客戶端的，則會誤解鎖          jedis.del(lockKey);      }  }

如果在釋放鎖的時候，這把鎖已經不屬於這個客戶端（例如已經過期，並且被別的客戶端獲取鎖成功了），那就會出現釋放了其他客戶端的鎖的情況。

​ 所以我們把判斷客戶端是否相等和刪除 key 的操作放在 Lua 腳本裡面執行。

/**       * 釋放分散式鎖       * @param jedis Redis客戶端       * @param lockKey 鎖       * @param requestId 請求標識       * @return 是否釋放成功       */      public static boolean releaseDistributedLock(Jedis jedis, String lockKey, String requestId) {          String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";          Object result = jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(requestId));            if (RELEASE_SUCCESS.equals(result)) {              return true;          }          return false;      }   

這個是 Jedis 裡面分散式鎖的實現。

1.3 Luttece

https://lettuce.io/

1.3.1 特點

與 Jedis 相比，Lettuce 則完全克服了其執行緒不安全的缺點：Lettuce 是一個可伸縮的執行緒安全的 Redis 客戶端，支援同步、非同步和響應式模式（Reactive）。多個執行緒可以共享一個連接實例，而不必擔心多執行緒並發問題。
同步調用：

public class LettuceSyncTest {      public static void main(String[] args) {          // 創建客戶端          RedisClient client = RedisClient.create("redis://127.0.0.1:6379");          // 執行緒安全的長連接，連接丟失時會自動重連          StatefulRedisConnection<String, String> connection = client.connect();          // 獲取同步執行命令，默認超時時間為 60s          RedisCommands<String, String> sync = connection.sync();          // 發送get請求，獲取值          sync.set("gupao:sync","lettuce-sync-666" );          String value = sync.get("gupao:sync");          System.out.println("------"+value);          //關閉連接          connection.close();          //關掉客戶端          client.shutdown();      }  }  

非同步的結果使用 RedisFuture 包裝，提供了大量回調的方法。
非同步調用：

  import io.lettuce.core.RedisClient;  import io.lettuce.core.RedisFuture;  import io.lettuce.core.api.StatefulRedisConnection;  import io.lettuce.core.api.async.RedisAsyncCommands;    import java.util.concurrent.ExecutionException;  import java.util.concurrent.TimeUnit;  import java.util.concurrent.TimeoutException;    public class LettuceASyncTest {      public static void main(String[] args) {          RedisClient client = RedisClient.create("redis://127.0.0.1:6379");          // 執行緒安全的長連接，連接丟失時會自動重連          StatefulRedisConnection<String, String> connection = client.connect();          // 獲取非同步執行命令api          RedisAsyncCommands<String, String> commands = connection.async();          // 獲取RedisFuture<T>          commands.set("gupao:async","lettuce-async-666");          RedisFuture<String> future = commands.get("gupao:async");          try {              String value = future.get(60, TimeUnit.SECONDS);              System.out.println("------"+value);          } catch (InterruptedException | ExecutionException | TimeoutException e) {              e.printStackTrace();          }        }  }

它基於 Netty 框架構建，支援 Redis 的高級功能，如 Pipeline、發布訂閱，事務、Sentinel，集群，支援連接池。

​ Lettuce 是 Spring Boot 2.x 默認的客戶端，替換了 Jedis。集成之後我們不需要單獨使用它，直接調用 Spring 的 RedisTemplate 操作，連接和創建和關閉也不需要我們操心。

<dependency>      <groupId>org.springframework.boot</groupId>      <artifactId>spring-boot-starter-data-redis</artifactId>  </dependency>

1.4 Redisson

https://redisson.org/
https://github.com/redisson/redisson/wiki/目錄

1.4.1 本質

Redisson 是一個在 Redis 的基礎上實現的 Java 駐記憶體數據網格（In-Memory
Data Grid），提供了分散式和可擴展的 Java 數據結構。

1.4.2 特點

基於 Netty 實現，採用非阻塞 IO，性能高
支援非同步請求
支援連接池、pipeline、LUA Scripting、Redis Sentinel、Redis Cluster
不支援事務，官方建議以 LUA Scripting 代替事務
主從、哨兵、集群都支援。Spring 也可以配置和注入 RedissonClient。

1.4.3 實現分散式鎖

在 Redisson 裡面提供了更加簡單的分散式鎖的實現。

public static void main(String[] args) throws InterruptedException {      RLock rLock=redissonClient.getLock("updateAccount");      // 最多等待 100 秒、上鎖 10s 以後自動解鎖      if(rLock.tryLock(100,10, TimeUnit.SECONDS)){          System.out.println("獲取鎖成功");      }      // do something      rLock.unlock();  }

在獲得 RLock 之後，只需要一個 tryLock 方法，裡面有 3 個參數：
1、watiTime：獲取鎖的最大等待時間，超過這個時間不再嘗試獲取鎖
2、leaseTime：如果沒有調用 unlock，超過了這個時間會自動釋放鎖
3、TimeUnit：釋放時間的單位

Redisson 的分散式鎖是怎麼實現的呢？
在加鎖的時候，在 Redis 寫入了一個 HASH，key 是鎖名稱，field 是執行緒名稱，value是 1（表示鎖的重入次數）

源碼：
tryLock()——tryAcquire()——tryAcquireAsync()——tryLockInnerAsync()

最終也是調用了一段 Lua 腳本。裡面有一個參數，兩個參數的值

// KEYS[1] 鎖名稱 updateAccount  // ARGV[1] key 過期時間 10000ms  // ARGV[2] 執行緒名稱  // 鎖名稱不存在  if (redis.call('exists', KEYS[1]) == 0) then      // 創建一個 hash，key=鎖名稱，field=執行緒名，value=1      redis.call('hset', KEYS[1], ARGV[2], 1);      // 設置 hash 的過期時間      redis.call('pexpire', KEYS[1], ARGV[1]);      return nil;  end;  // 鎖名稱存在，判斷是否當前執行緒持有的鎖  if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then      // 如果是，value+1，代表重入次數+1      redis.call('hincrby', KEYS[1], ARGV[2], 1);      // 重新獲得鎖，需要重新設置 Key 的過期時間      redis.call('pexpire', KEYS[1], ARGV[1]);      return nil;  end;  // 鎖存在，但是不是當前執行緒持有，返回過期時間（毫秒）  return redis.call('pttl', KEYS[1])

釋放鎖，源碼：

unlock——unlockInnerAsync

// KEYS[1] 鎖的名稱 updateAccount  // KEYS[2] 頻道名稱 redisson_lock__channel:{updateAccount}  // ARGV[1] 釋放鎖的消息 0  // ARGV[2] 鎖釋放時間 10000  // ARGV[3] 執行緒名稱  // 鎖不存在（過期或者已經釋放了）  if (redis.call('exists', KEYS[1]) == 0) then      // 發布鎖已經釋放的消息      redis.call('publish', KEYS[2], ARGV[1]);      return 1;  end;  // 鎖存在，但是不是當前執行緒加的鎖  if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then      return nil;  end;  ​  // 鎖存在，是當前執行緒加的鎖  // 重入次數-1  local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1);  // -1 後大於 0，說明這個執行緒持有這把鎖還有其他的任務需要執行  if (counter > 0) then      // 重新設置鎖的過期時間      redis.call('pexpire', KEYS[1], ARGV[2]);      return 0;  else      // -1 之後等於 0，現在可以刪除鎖了      redis.call('del', KEYS[1]);      // 刪除之後發布釋放鎖的消息      redis.call('publish', KEYS[2], ARGV[1]);      return 1;  end;  ​  // 其他情況返回 nil  return nil;

​ 這個是 Redisson 裡面分散式鎖的實現，我們在調用的時候非常簡單。
​ Redisson 跟 Jedis 定位不同，它不是一個單純的 Redis 客戶端，而是基於 Redis 實現的分散式的服務，如果有需要用到一些分散式的數據結構，比如我們還可以基於Redisson 的分散式隊列實現分散式事務，就可以引入 Redisson 的依賴實現。

2 數據 一致性

2.1 快取 使用場景

針對讀多寫少的高並發場景，我們可以使用快取來提升查詢速度。
當我們使用 Redis 作為快取的時候，一般流程是這樣的：
1、如果數據在 Redis 存在，應用就可以直接從 Redis 拿到數據，不用訪問資料庫。

2、如果 Redis 裡面沒有，先到資料庫查詢，然後寫入到 Redis，再返回給應用

2.2 一致性 問題的定義

因為這些數據是很少修改的，所以在絕大部分的情況下可以命中快取。但是，一旦被快取的數據發生變化的時候，我們既要操作資料庫的數據，也要操作 Redis 的數據，所以問題來了。現在我們有兩種選擇：

​ 1、先操作 Redis 的數據再操作資料庫的數據
​ 2、先操作資料庫的數據再操作 Redis 的數據

​ 到底選哪一種？
​ 首先需要明確的是，不管選擇哪一種方案， 我們肯定是希望兩個操作要麼都成功，要麼都一個都不成功。不然就會發生 Redis 跟資料庫的數據不一致的問題。

​ 但是，Redis 的數據和資料庫的數據是不可能通過事務達到統一的，我們只能根據相應的場景和所需要付出的代價來採取一些措施降低數據不一致的問題出現的概率，在數據一致性和性能之間取得一個權衡。
​ 對於資料庫的實時性一致性要求不是特別高的場合，比如 T+1 的報表，可以採用定時任務查詢資料庫數據同步到 Redis 的方案。
​ 由於我們是以資料庫的數據為準的，所以給快取設置一個過期時間，是保證最終一致性的解決方案。

2.3 方案 選擇

2.3.1 Redis ：刪除還是 更新？

​ 這裡我們先要補充一點，當存儲的數據發生變化，Redis 的數據也要更新的時候，我們有兩種方案，一種就是直接更新，調用 set；還有一種是直接刪除快取，讓應用在下次查詢的時候重新寫入。
​ 這兩種方案怎麼選擇呢？這裡我們主要考慮更新快取的代價。
​ 更新快取之前，是不是要經過其他表的查詢、介面調用、計算才能得到最新的數據，而不是直接從資料庫拿到的值。如果是的話，建議直接刪除快取，這種方案更加簡單，而且避免了資料庫的數據和快取不一致的情況。在一般情況下，我們也推薦使用刪除的方案。

這一點明確之後，現在我們就剩一個問題：

​ 1、到底是先更新資料庫，再刪除快取

​ 2、還是先刪除快取，再更新資料庫

我們先看第一種方案。

2.3.2 先更新 資料庫，再刪除快取

​ 正常情況：
​ 更新資料庫，成功。
​ 刪除快取，成功。

​ 異常情況：
​ 1、更新資料庫失敗，程式捕獲異常，不會走到下一步，所以數據不會出現不一致。
​ 2、更新資料庫成功，刪除快取失敗。資料庫是新數據，快取是舊數據，發生了不一致的情況。
​ 這種問題怎麼解決呢？我們可以提供一個重試的機制。
​ 比如：如果刪除快取失敗，我們捕獲這個異常，把需要刪除的 key 發送到消息隊列。讓後自己創建一個消費者消費，嘗試再次刪除這個 key。
​ 這種方式有個缺點，會對業務程式碼造成入侵。

​ 所以我們又有了第二種方案（非同步更新快取）：
​ 因為更新資料庫時會往 binlog 寫入日誌，所以我們可以通過一個服務來監聽 binlog的變化（比如阿里的 canal），然後在客戶端完成刪除 key 的操作。如果刪除失敗的話，再發送到消息隊列。
​ 總之，對於後刪除快取失敗的情況，我們的做法是不斷地重試刪除，直到成功。
​ 無論是重試還是非同步刪除，都是最終一致性的思想。

2.3.3 先 刪除快取，再更新資料庫

​ 正常情況：
​ 刪除快取，成功。
​ 更新資料庫，成功。

​ 異常情況：
​ 1、刪除快取，程式捕獲異常，不會走到下一步，所以數據不會出現不一致。
​ 2、刪除快取成功，更新資料庫失敗。 因為以資料庫的數據為準，所以不存在數據不一致的情況。

​ 看起來好像沒問題，但是如果有程式並發操作的情況下：
​ 1）執行緒 A 需要更新數據，首先刪除了 Redis 快取
​ 2）執行緒 B 查詢數據，發現快取不存在，到資料庫查詢舊值，寫入 Redis，返回
​ 3）執行緒 A 更新了資料庫

​ 這個時候，Redis 是舊的值，資料庫是新的值，發生了數據不一致的情況。

​ 那問題就變成了：能不能讓對同一條數據的訪問串列化呢？程式碼肯定保證不了，因為有多個執行緒，即使做了任務隊列也可能有多個服務實例。資料庫也保證不了，因為會有多個資料庫的連接。只有一個資料庫只提供一個連接的情況下，才能保證讀寫的操作是串列的，或者我們把所有的讀寫請求放到同一個記憶體隊列當中，但是這種情況吞吐量太低了。

​ 所以我們有一種延時雙刪的策略，在寫入數據之後，再刪除一次快取。

​ A 執行緒：
​ 1）刪除快取
​ 2）更新資料庫
​ 3）休眠 500ms（這個時間，依據讀取數據的耗時而定）
​ 4）再次刪除快取

偽程式碼：

public void write(String key,Object data){      redis.delKey(key);      db.updateData(data);      Thread.sleep(500);      redis.delKey(key);  }

3 高並發 問題

​ 在 Redis 存儲的所有數據中，有一部分是被頻繁訪問的。有兩種情況可能會導致熱點問題的產生，一個是用戶集中訪問的數據，比如搶購的商品，明星結婚和明星出軌的微博。還有一種就是在數據進行分片的情況下，負載不均衡，超過了單個伺服器的承受能力。熱點問題可能引起快取服務的不可用，最終造成壓力堆積到資料庫。

​ 出於存儲和流量優化的角度，我們必須要找到這些熱點數據。

3.1 熱點數據 發現

​ 除了自動的快取淘汰機制之外，怎麼找出那些訪問頻率高的 key 呢？或者說，我們可以在哪裡記錄 key 被訪問的情況呢？

3.1.1 客戶端

​ 第一個當然是在客戶端了，比如我們可不可以在所有調用了 get、set 方法的地方，加上 key 的計數。但是這樣的話，每一個地方都要修改，重複的程式碼也多。如果我們用的是 Jedis 的客戶端，我們可以在 Jedis 的 Connection 類的 sendCommand()裡面，用一個 HashMap 進行 key 的計數。

​ 但是這種方式有幾個問題：
​ 1、不知道要存多少個 key，可能會發生記憶體泄露的問題。
​ 2、會對客戶端的程式碼造成入侵。
​ 3、只能統計當前客戶端的熱點 key。

3.1.2 代理層

​ 第二種方式就是在代理端實現，比如 TwemProxy 或者 Codis，但是不是所有的項目都使用了代理的架構。

3.1.3 服務端

​ 第三種就是在服務端統計，Redis 有一個 monitor 的命令，可以監控到所有 Redis執行的命令。

程式碼：

jedis.monitor(new JedisMonitor() {      @Override      public void onCommand(String command) {          System.out.println("#monitor: " + command);      }  }); 

Facebook 的 開 源 項 目 redis-faina（https://github.com/facebookarchive/redis-faina.git）就是基於這個原理實現的。它是一個 python 腳本，可以分析 monitor 的數據。

redis-cli -p 6379 monitor | head -n 100000 | ./redis-faina.py

​ 這種方法也會有兩個問題：1）monitor 命令在高並發的場景下，會影響性能，所以不適合長時間使用。
​ 只能統計一個 Redis 節點的熱點 key。

3.1.4 機器層面

​ 還有一種方法就是機器層面的，通過對 TCP 協議進行抓包，也有一些開源的方案，
比如 ELK 的 packetbeat 插件。

​ 當我們發現了熱點 key 之後，我們來看下熱點數據在高並發的場景下可能會出現的
問題，以及怎麼去解決。

3.2 快取 雪崩

3.2.1 什麼 是快取 雪崩

​ 快取雪崩就是 Redis 的大量熱點數據同時過期（失效），因為設置了相同的過期時間，剛好這個時候 Redis 請求的並發量又很大，就會導致所有的請求落到資料庫。

3.2.2 快取雪崩 的解決方案

​ 1）加互斥鎖或者使用隊列，針對同一個 key 只允許一個執行緒到資料庫查詢
​ 2）快取定時預先更新，避免同時失效
​ 3）通過加隨機數，使 key 在不同的時間過期
​ 4）快取永不過期

3.3 快取穿透

3.3.1 快取 穿透 何時 發生

我們已經知道了 Redis 使用的場景了。在快取存在和快取不存在的情況下的什麼情況我們都了解了。

​ 還有一種情況，數據在資料庫和 Redis 裡面都不存在，可能是一次條件錯誤的查詢。在這種情況下，因為資料庫值不存在，所以肯定不會寫入 Redis，那麼下一次查詢相同的key 的時候，肯定還是會再到資料庫查一次。那麼這種循環查詢資料庫中不存在的值，並且每次使用的是相同的 key 的情況，我們有沒有什麼辦法避免應用到資料庫查詢呢？

（1）快取空數據 （2）快取特殊字元串，比如&&
我們可以在資料庫快取一個空字元串，或者快取一個特殊的字元串，那麼在應用裡面拿到這個特殊字元串的時候，就知道資料庫沒有值了，也沒有必要再到資料庫查詢了。但是這裡需要設置一個過期時間，不然的話資料庫已經新增了這一條記錄，應用也還是拿不到值。

​ 這個是應用重複查詢同一個不存在的值的情況，如果應用每一次查詢的不存在的值是不一樣的呢？即使你每次都快取特殊字元串也沒用，因為它的值不一樣，比如我們的用戶系統登錄的場景，如果是惡意的請求，它每次都生成了一個符合 ID 規則的帳號，但是這個帳號在我們的資料庫是不存在的，那 Redis 就完全失去了作用

​ 這種因為每次查詢的值都不存在導致的 Redis 失效的情況，我們就把它叫做快取穿透。這個問題我們應該怎麼去解決呢？

3.3.2 經典 面試題

​ 其實它也是一個通用的問題，關鍵就在於我們怎麼知道請求的 key 在我們的資料庫裡面是否存在，如果數據量特別大的話，我們怎麼去快速判斷。

​ 這也是一個非常經典的面試題：
​ 如何在海量元素中（例如 10 億無序、不定長、不重複）快速判斷一個元素是否存在？

​ 如果是快取穿透的這個問題，我們要避免到資料庫查詢不存的數據，肯定要把這 10億放在別的地方。這些數據在 Redis 裡面也是沒有的，為了加快檢索速度，我們要把數據放到記憶體裡面來判斷，問題來了：
​ 如果我們直接把這些元素的值放到基本的數據結構（List、Map、Tree）裡面，比如一個元素 1 位元組的欄位，10 億的數據大概需要 900G 的記憶體空間，這個對於普通的伺服器來說是承受不了的。
​ 所以，我們存儲這幾十億個元素，不能直接存值，我們應該找到一種最簡單的最節省空間的數據結構，用來標記這個元素有沒有出現。
​ 這個東西我們就把它叫做點陣圖，他是一個有序的數組，只有兩個值，0 和 1。0 代表不存在，1 代表存在。

​ 那我們怎麼用這個數組裡面的有序的位置來標記這10億個元素是否存在呢？我們是不是必須要有一個映射方法，把元素映射到一個下標位置上？
​ 對於這個映射方法，我們有幾個基本的要求：
​ 1）因為我們的值長度是不固定的，我希望不同長度的輸入，可以得到固定長度的輸出。
​ 2）轉換成下標的時候，我希望他在我的這個有序數組裡面是分布均勻的，不然的話全部擠到一對去了，我也沒法判斷到底哪個元素存了，哪個元素沒存。
​ 這個就是哈希函數，比如 MD5、SHA-1 等等這些都是常見的哈希演算法。

比如，這 6 個元素，我們經過哈希函數和位運算，得到了相應的下標。

3.3.3 哈希碰撞

​ 這個時候，Tom 和 Mic 經過計算得到的哈希值是一樣的，那麼再經過位運算得到的下標肯定是一樣的，我們把這種情況叫做哈希衝突或者哈希碰撞。
​ 如果發生了哈希碰撞，這個時候對於我們的容器存值肯定是有影響的，我們可以通過哪些方式去降低哈希碰撞的概率呢？
​ 第一種就是擴大維數組的長度或者說點陣圖容量。因為我們的函數是分布均勻的，所以，點陣圖容量越大，在同一個位置發生哈希碰撞的概率就越小。
​ 是不是點陣圖容量越大越好呢？不管存多少個元素，都創建一個幾萬億大小的點陣圖，可以嗎？當然不行，因為越大的點陣圖容量，意味著越多的記憶體消耗，所以我們要創建一個合適大小的點陣圖容量。
​ 除了擴大點陣圖容量，我們還有什麼降低哈希碰撞概率的方法呢？
​ 如果兩個元素經過一次哈希計算，得到的相同下標的概率比較高，我可以不可以計算多次呢？ 原來我只用一個哈希函數，現在我對於每一個要存儲的元素都用多個哈希函數計算，這樣每次計算出來的下標都相同的概率就小得多了。
​ 同樣的，我們能不能引入很多個哈希函數呢？比如都計算 100 次，都可以嗎？當然也會有問題，第一個就是它會填滿點陣圖的更多空間，第二個是計算是需要消耗時間的。
​ 所以總的來說，我們既要節省空間，又要很高的計算效率，就必須在點陣圖容量和函數個數之間找到一個最佳的平衡。
​ 比如說：我們存放 100 萬個元素，到底需要多大的點陣圖容量，需要多少個哈希函數呢？

3.3.4 布隆過濾器原理

​ 當然，這個事情早就有人研究過了，在 1970 年的時候，有一個叫做布隆的前輩對於判斷海量元素中元素是否存在的問題進行了研究，也就是到底需要多大的點陣圖容量和多少個哈希函數，它發表了一篇論文，提出的這個容器就叫做布隆過濾器。

​ 我們來看一下布隆過濾器的工作原理。
​ 首先，布隆過濾器的本質就是我們剛才分析的，一個位數組，和若干個哈希函數。

​ 集合裡面有 3 個元素，要把它存到布隆過濾器裡面去，應該怎麼做？首先是 a 元素，這裡我們用 3 次計算。b、c 元素也一樣。
​ 元素已經存進去之後，現在我要來判斷一個元素在這個容器裡面是否存在，就要使用同樣的三個函數進行計算。
​ 比如 d 元素，我用第一個函數 f1 計算，發現這個位置上是 1，沒問題。第二個位置也是 1，第三個位置也是 1 。
​ 如果經過三次計算得到的下標位置值都是 1，這種情況下，能不能確定 d 元素一定在這個容器裡面呢？ 實際上是不能的。比如這張圖裡面，這三個位置分別是把 a,b,c 存進去的時候置成 1 的，所以即使 d 元素之前沒有存進去，也會得到三個 1，判斷返回 true。

所以，這個是布隆過濾器的一個很重要的特性，因為哈希碰撞不可避免，所以它會存在一定的誤判率。這種把本來不存在布隆過濾器中的元素誤判為存在的情況，我們把它叫做假陽性（False Positive Probability，FPP）。
我們再來看另一個元素，e 元素。我們要判斷它在容器裡面是否存在，一樣地要用這三個函數去計算。第一個位置是 1，第二個位置是 1，第三個位置是 0。
e 元素是不是一定不在這個容器裡面呢？ 可以確定一定不存在。如果說當時已經把e 元素存到布隆過濾器裡面去了，那麼這三個位置肯定都是 1，不可能出現 0。
總結：布隆過濾器的特點：
從容器的角度來說：
1、如果布隆過濾器判斷元素在集合中存在，不一定存在
2、如果布隆過濾器判斷不存在，一定不存在從元素的角度來說：
3、如果元素實際存在，布隆過濾器一定判斷存在
4、如果元素實際不存在，布隆過濾器可能判斷存在利用，第二個特性，我們是不是就能解決持續從資料庫查詢不存在的值的問題？

3.3.5 Guava 的 實現

Google的 Guava 裡面就提供了一個現成的布隆過濾器。

<dependency>      <groupId>com.google.guava</groupId>      <artifactId>guava</artifactId>      <version>21.0</version>  </dependency>

創建布隆過濾器：

BloomFilter<String> bf = BloomFilter.create(Funnels.stringFunnel(Charsets.UTF_8), insertions);

布隆過濾器提供的存放元素的方法是 put()。
布隆過濾器提供的判斷元素是否存在的方法是 mightContain()。

if (bf.mightContain(data)) {      if (sets.contains(data)) {      // 判斷存在實際存在的時候，命中      right++;      continue;      }      // 判斷存在卻不存在的時候，錯誤      wrong++;  }

布隆過濾器把誤判率默認設置為 0.03，也可以在創建的時候指定。

public static <T> BloomFilter<T> create(Funnel<? super T> funnel, long expectedInsertions) {      return create(funnel, expectedInsertions, 0.03D);  }

點陣圖的容量是基於元素個數和誤判率計算出來的。

long numBits = optimalNumOfBits(expectedInsertions, fpp);

根據位數組的大小，我們進一步計算出了哈希函數的個數。

int numHashFunctions = optimalNumOfHashFunctions(expectedInsertions, numBits);

存儲 100 萬個元素只佔用了 0.87M 的記憶體，生成了 5 個哈希函數。

https://hur.st/bloomfilter/?n=1000000&p=0.03&m=&k=

3.3.6 布隆過濾器 在項目中的使用

布隆過濾器的工作位置：

因為要判斷資料庫的值是否存在，所以第一步是載入資料庫所有的數據。在去 Redis查詢之前，先在布隆過濾器查詢，如果 bf 說沒有，那資料庫肯定沒有，也不用去查了。如果 bf 說有，才走之前的流程。

import com.google.common.base.Charsets;  import com.google.common.hash.BloomFilter;  import com.google.common.hash.Funnels;  import com.gupaoedu.entity.User;  import com.gupaoedu.service.UserService;  import org.junit.Test;  import org.junit.runner.RunWith;  import org.springframework.beans.factory.annotation.Autowired;  import org.springframework.boot.autoconfigure.EnableAutoConfiguration;  import org.springframework.boot.test.context.SpringBootTest;  import org.springframework.data.redis.core.RedisTemplate;  import org.springframework.data.redis.core.ValueOperations;  import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;    import javax.annotation.PostConstruct;  import javax.annotation.Resource;  import java.text.SimpleDateFormat;  import java.util.Date;  import java.util.List;  import java.util.UUID;  import java.util.concurrent.BrokenBarrierException;  import java.util.concurrent.CyclicBarrier;  import java.util.concurrent.ExecutorService;  import java.util.concurrent.Executors;      @RunWith(SpringJUnit4ClassRunner.class)  @SpringBootTest  @EnableAutoConfiguration  public class BloomTestsConcurrency {      @Resource      private RedisTemplate redisTemplate;        @Autowired      private UserService userService;        private static final int THREAD_NUM = 1000; // 並發執行緒數量，Windows機器不要設置過大        static BloomFilter<String> bf;        static List<User> allUsers;        @PostConstruct      public void init() {          // 從資料庫獲取數據，載入到布隆過濾器          long start = System.currentTimeMillis();          allUsers = userService.getAllUser();          if (allUsers == null || allUsers.size() == 0) {              return;          }          // 創建布隆過濾器，默認誤判率0.03，即3%          bf = BloomFilter.create(Funnels.stringFunnel(Charsets.UTF_8), allUsers.size());          // 誤判率越低，數組長度越長，需要的哈希函數越多          // bf = BloomFilter.create(Funnels.stringFunnel(Charsets.UTF_8), allUsers.size(), 0.0001);          // 將數據存入布隆過濾器          for (User user : allUsers) {              bf.put(user.getAccount());          }          long end = System.currentTimeMillis();          System.out.println("查詢並載入"+allUsers.size()+"條數據到布隆過濾器完畢，總耗時："+(end -start ) +"毫秒");      }        @Test      public void cacheBreakDownTest() {          long start = System.currentTimeMillis();          allUsers = userService.getAllUser();          CyclicBarrier cyclicBarrier = new CyclicBarrier(THREAD_NUM);          ExecutorService executorService = Executors.newFixedThreadPool(THREAD_NUM);          for (int i = 0; i < THREAD_NUM; i++){              executorService.execute(new BloomTestsConcurrency().new MyThread(cyclicBarrier, redisTemplate, userService));          }            executorService.shutdown();          //判斷是否所有的執行緒已經運行完          while (!executorService.isTerminated()) {            }            long end = System.currentTimeMillis();          System.out.println("並發數："+THREAD_NUM + "，新建執行緒以及過濾總耗時："+(end -start ) +"毫秒，演示結束");      }        public class MyThread implements Runnable {          private CyclicBarrier cyclicBarrier;          private RedisTemplate redisTemplate;          private UserService userService;            public MyThread(CyclicBarrier cyclicBarrier, RedisTemplate redisTemplate, UserService userService) {              this.cyclicBarrier = cyclicBarrier;              this.redisTemplate = redisTemplate;              this.userService = userService;          }            @Override          public void run() {              //所有子執行緒等待，當子執行緒全部創建完成再一起並發執行後面的程式碼              try {                  cyclicBarrier.await();              } catch (InterruptedException e) {                  e.printStackTrace();              } catch (BrokenBarrierException e) {                  e.printStackTrace();              }                // 1.1 （測試：布隆過濾器判斷不存在，攔截——如果沒有布隆過濾器，將造成快取穿透）              // 隨機產生一個字元串，在布隆過濾器中不存在              String randomUser = UUID.randomUUID().toString();              // 1.2 （測試：布隆過濾器判斷存在，從Redis快取取值，如果Redis為空則查詢資料庫並寫入Redis）              // 從List中獲取一個存在的用戶              // String randomUser = allUsers.get(new Random().nextInt(allUsers.size())).getAccount();              String key = "Key:" + randomUser;                Date date1 = new Date();              SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");                // 如果布隆過濾器中不存在這個用戶直接返回，將流量擋掉  /*            if (!bf.mightContain(randomUser)) {                  System.out.println(sdf.format(date1)+" 布隆過濾器中不存在，非法請求");                  return;              }*/                // 查詢快取，如果快取中存在直接返回快取數據              ValueOperations<String, String> operation =                      (ValueOperations<String, String>) redisTemplate.opsForValue();              Object cacheUser = operation.get(key);              if (cacheUser != null) {                  Date date2 = new Date();                  System.out.println(sdf.format(date2)+" 命中redis快取");                  return;              }                // TODO 防止並發重複寫快取，加鎖              synchronized (randomUser) {                  // 如果快取不存在查詢資料庫                  List<User> user = userService.getUserByAccount(randomUser);                  if (user == null || user.size() == 0) {                      // 很容易發生連接池不夠用的情況 HikariPool-1 - Connection is not available, request timed out after                      System.out.println(" Redis快取不存在，查詢資料庫也不存在，發生快取穿透！！！");                      return;                  }                  // 將mysql資料庫查詢到的數據寫入到redis中                  Date date3 = new Date();                  System.out.println(sdf.format(date3)+" 從資料庫查詢並寫入Reids");                  operation.set("Key:" + user.get(0).getAccount(), user.get(0).getAccount());              }          }        }      }

3.3.7 布隆過濾器 的其他應用場景

​ 布隆過濾器解決的問題是什麼？如何在海量元素中快速判斷一個元素是否存在。所以除了解決快取穿透的問題之外，我們還有很多其他的用途。
​ 比如爬數據的爬蟲，爬過的 url 我們不需要重複爬，那麼在幾十億的 url 裡面，怎麼判斷一個 url 是不是已經爬過了？
​ 還有我們的郵箱伺服器，發送垃圾郵件的帳號我們把它們叫做 spamer，在這麼多的郵箱帳號裡面，怎麼判斷一個帳號是不是 spamer 等等一些場景，我們都可以用到布隆過濾器。

公眾號

如果大家想要實時關注我更新的文章以及分享的乾貨的話，可以關注我的公眾號。