Redis實戰篇

1 Redis 客戶端

1.1 客戶端通信 原理

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

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

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

*3\r\n$3\r\nSET\r\n$8\r\nqingshan\r\n$4\r\n2673\r\n

get qingshan 抓包：

*2\r\n$3\r\nGET\r\n$8\r\nqingshan\r\n

​ 客戶端跟 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。

客戶端	描述
Jedis	A blazingly small and sane redis java client
lettuce	Advanced Redis client for thread-safe sync, async, and reactive usage. Supports Cluster, Sentinel,Pipelining, and codecs
Redisson	distributed and scalable Java data structures on top of Redis server

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

配置	作用
class	客戶端類型，分為三種。a）normal：普通客戶端；b）slave：slave 客戶端，用於復制；c）pubsub：發布訂閱客戶端
hard limit	如果客戶端使用的輸出緩沖區大於 ，客戶端會被立即關閉，0 代表不限制
soft limit soft seconds	如果客戶端使用的輸出緩沖區超過了 並且持續了 秒，客戶端會被立即關閉

每個客戶端使用的輸出緩沖區的大小可以用 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]	getName()	鎖的名稱（key）	updateAccount
ARGV[1]	internalLockLeaseTime	鎖釋放時間（毫秒)	10000
ARGV[2]	getLockName(threadId)	線程名稱	b60a9c8c-92f8-4bfe-b0e7-308967346336:1

// 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]	getName()	鎖名稱	updateAccount
KEYS[2]	getChannelName()	頻道名稱	redisson_lock__channel:{updateAccount}
KEYS[3]	LockPubSub.unlockMessage	解鎖時的消息	0
KEYS[4]	internalLockLeaseTime	釋放鎖的時間	10000
KEYS[5]	getLockName(threadId)	線程名稱	b60a9c8c-92f8-4bfe-b0e7-308967346336:1

// 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 的 實現

谷歌的 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 等等一些場景，我們都可以用到布隆過濾器。

公眾號

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