一、使用場(chǎng)景

試想一個(gè)場(chǎng)景，有一個(gè)配置服務(wù)系統(tǒng)，里面存儲(chǔ)著各種各樣的配置，比如直播間的直播信息、點(diǎn)贊、簽到、紅包、帶貨等等。這些配置信息有兩個(gè)特點(diǎn)：

并發(fā)量可能會(huì)特別特別大，試想一下，一個(gè)幾十萬(wàn)人的直播間，可能在直播開始前幾秒鐘，用戶就瞬間涌入進(jìn)來(lái)了，那么這時(shí)候我們的系統(tǒng)就得加載這些配置信息。此時(shí)請(qǐng)求量就如同洪峰一般，一下子就沖擊進(jìn)入我們的系統(tǒng)。

這些配置通常都是只需要讀取，在B端（管理后臺(tái)）設(shè)置好的，一般直播開始后，修改的頻率很低。

那么面對(duì)上述的業(yè)務(wù)場(chǎng)景，假設(shè)我們的目標(biāo)是扛住3wQPS，你們會(huì)選用什么技術(shù)架構(gòu)和方案呢？

1、直接查數(shù)據(jù)庫(kù)，例如MySQL、Doris之類的關(guān)系型數(shù)據(jù)庫(kù)。很明顯這肯定扛不住，一般關(guān)系型數(shù)據(jù)庫(kù)能讓扛?jìng)€(gè)幾千就基本上到頭了。

2、使用單機(jī)版Redis。理論上是可以的，騰訊云（下圖）和一些Redis官方的數(shù)據(jù)，都說(shuō)理論上高配置版本的單機(jī)Redis能抗住10W+的QPS?？墒抢碚摦吘故抢碚摚瑢?shí)際上工作中，我使用Redis做過(guò)許多壓測(cè)，都表明單機(jī)Redis上了兩萬(wàn)多之后就性能會(huì)出現(xiàn)瓶頸，壓測(cè)就壓不上去。（當(dāng)然，或許是我司的Redis還沒(méi)升到頂配？）

3、使用集群版Redis，當(dāng)然是可以解決這個(gè)問(wèn)題，就是成本有點(diǎn)點(diǎn)高咯，公司不差錢完全可以使用這個(gè)方案。

4、本地緩存，就是本文的重點(diǎn)，完美地解決這個(gè)問(wèn)題。所謂本地緩存就是將這些所需要獲取的數(shù)據(jù)存儲(chǔ)在服務(wù)器的內(nèi)存中。服務(wù)器讀取本地緩存的速度理論上來(lái)說(shuō)沒(méi)有上限，看服務(wù)器物理機(jī)的配置。但其下限就遠(yuǎn)比MySQL和單機(jī)Redis之類的高好幾倍了，一臺(tái)2核4G的Linux服務(wù)器，估計(jì)也至少10W+QPS起步。我曾經(jīng)在本地的Windows系統(tǒng)做過(guò)壓測(cè)（四核八線程，16G），就達(dá)到過(guò)100W+的QPS。換到同等配置的Linux系統(tǒng)上，那就更不用說(shuō)了。

二、技術(shù)方案

既然選用了本地緩存這個(gè)策略，那么我們?cè)趺丛O(shè)計(jì)這個(gè)本地緩存的技術(shù)方案呢？

1、如上圖所示，我們客戶端獲取數(shù)據(jù)首先會(huì)讀取本地緩存，如果本地緩存沒(méi)有數(shù)據(jù)就會(huì)讀取Redis數(shù)據(jù)，如果Redis沒(méi)有就會(huì)讀取DB數(shù)據(jù)。

2、需要注意的是，本地緩存和DB之間一般還會(huì)加入Redis這一層緩存。這是因?yàn)楸镜鼐彺嬖O(shè)置好后就無(wú)法再更新了（除非重啟服務(wù)器），而Redis緩存我們是可以在DB有更改后，隨時(shí)更新。這個(gè)也很好理解，因?yàn)镽edis是有單獨(dú)的Redis服務(wù)器，而本地緩存就只能在那臺(tái)機(jī)器上更新和設(shè)置，但實(shí)際項(xiàng)目中，設(shè)置本地緩存的DB數(shù)據(jù)源的機(jī)器和使用本地緩存的機(jī)器大概率都不在同一個(gè)系統(tǒng)中。所以我們本地緩存的時(shí)間都設(shè)置得很短，大部分都是秒級(jí)的，一般不會(huì)超過(guò)1分鐘，比如1秒、2秒... 。而Redis這個(gè)緩存時(shí)長(zhǎng)明顯可以設(shè)置長(zhǎng)一些，比如半小時(shí)、1小時(shí)...。

三、如何更新本地緩存

上面講了，本地緩存最不好的地方就是更新問(wèn)題，因?yàn)楹芸赡茉O(shè)置本地緩存的DB數(shù)據(jù)源的系統(tǒng)和使用本地緩存的系統(tǒng)不是同一個(gè)，無(wú)法在DB數(shù)據(jù)更新的時(shí)候就同步更新本地緩存。但是實(shí)際使用的時(shí)候很可能就需要這種場(chǎng)景，就是在更新數(shù)據(jù)源的時(shí)候去更新本地緩存。舉個(gè)例子：

我們?cè)O(shè)置配置A的DB數(shù)據(jù)源的系統(tǒng)是一個(gè)API系統(tǒng)，但現(xiàn)在有一個(gè)腳本系統(tǒng)，需要根據(jù)某個(gè)配置A，去處理C端的一些行為數(shù)據(jù)，判斷是否滿足該配置A，然后進(jìn)行對(duì)應(yīng)的業(yè)務(wù)處理。好了，現(xiàn)在C端的行為數(shù)據(jù)量是非常龐大的，可以說(shuō)是海量數(shù)據(jù)，平均每秒鐘有五十萬(wàn)的數(shù)據(jù)通過(guò)kafka推送過(guò)來(lái)。此時(shí)我們就必須得用本地緩存存儲(chǔ)配置A的信息了，才能抗的住這個(gè)流量洪峰。但是這是一個(gè)腳本系統(tǒng)啊，我們更新配置A的DB信息是在對(duì)應(yīng)的API系統(tǒng)中的。那怎么辦呢？

有幾個(gè)方法：

1、在腳本系統(tǒng)中維護(hù)一個(gè)腳本，每隔一段時(shí)間就去讀取MySQL的數(shù)據(jù)，然后更新到本地緩存。但這個(gè)得綜合評(píng)估下時(shí)間和MySQL的性能，因?yàn)橐恢睊弑怼?/p>

2、拉取MySQL的binlog日志，每當(dāng)數(shù)據(jù)有變更時(shí)，kafka推送數(shù)據(jù)到下游。腳本監(jiān)聽kafka數(shù)據(jù)，當(dāng)收到kafka數(shù)據(jù)是就更新配置A的本地緩存。但這個(gè)也得注意，因?yàn)槟_本系統(tǒng)一般會(huì)同時(shí)起很多個(gè)服務(wù)，所以得注意有多少個(gè)服務(wù)就得設(shè)置多少個(gè)消費(fèi)者組，因?yàn)橐ＷC腳本系統(tǒng)的每個(gè)服務(wù)都消費(fèi)到kafka對(duì)應(yīng)的DB更新數(shù)據(jù)，進(jìn)而更新各自機(jī)器上的本地緩存。

3、使用Redis的發(fā)布訂閱功能，上游api有更新配置信息時(shí)就去發(fā)布信息，每個(gè)腳本服務(wù)都去訂閱該信息，一有消息就去更新自己機(jī)器上的本地緩存。但這也有個(gè)弊端，Redis的發(fā)布訂閱功能是沒(méi)有確認(rèn)機(jī)制的，所以可能某個(gè)腳本服務(wù)沒(méi)收到信息導(dǎo)致沒(méi)更新本地緩存，然后就出現(xiàn)bug了。demo如下：

（1）發(fā)布者：

package main

import (
	"context"
	"fmt"
	"github.com/go-redis/redis/v8"
	"time"
)

var ctx = context.Background()

// 發(fā)布訂閱功能
// 發(fā)布者發(fā)布，所有訂閱者都能接收到發(fā)布的消息。注意區(qū)分消息隊(duì)列，消息隊(duì)列是發(fā)布者發(fā)布，只有一個(gè)訂閱者能搶到。
func main() {
	rdb := redis.NewClient(&redis.Options{
		Addr:     "localhost:16379",
		Password: "123456",
	})

	i := 0
	for {
		// 模擬數(shù)據(jù)更新時(shí)發(fā)布消息
		rdb.Publish(ctx, "money_updates", "New money value updated "+fmt.Sprintf("%d", i))
		fmt.Println("Message published " + fmt.Sprintf("%d", i))
		time.Sleep(5 * time.Second)
		i++
	}

}

（2）訂閱者：

package main

import (
	"context"
	"fmt"
	"github.com/go-redis/redis/v8"
)

var ctx = context.Background()

func main() {
	rdb := redis.NewClient(&redis.Options{
		Addr:     "localhost:16379",
		Password: "123456",
	})

	pubsub := rdb.Subscribe(ctx, "money_updates")
	defer pubsub.Close()

	// 等待消息
	for {
		msg, err := pubsub.ReceiveMessage(ctx)
		if err != nil {
			fmt.Println("Error receiving message:", err)
			return
		}
		fmt.Println("Received message:", msg.Payload)
	}
}

4、跟方法1類似，只是可以把修改的配置A信息推送到Redis中，然后腳本去掃描Redis信息，有則更新本地緩存。其實(shí)就是延遲隊(duì)列。但這個(gè)就得上游的配置A增刪改都要寫入這個(gè)Redis，有時(shí)候增刪改的口子太多，其實(shí)實(shí)施起來(lái)也比較困難。

如上所述，基本上更新本地緩存沒(méi)有一個(gè)很合適、很高效的方法，只能選取其中一個(gè)比較符合自己業(yè)務(wù)場(chǎng)景的方法。

四、本地緩存常用類庫(kù)

go如何使用本地緩存呢？

1、可以自己實(shí)現(xiàn)一個(gè)本地緩存，一般可以使用LRU（最近最少使用）算法。下面是自己實(shí)現(xiàn)的一個(gè)本地緩存的demo。

package main

import (
	"container/list"
	"fmt"
	"sync"
	"time"
)

type Cache struct {
	capacity int
	cache    map[int]*list.Element
	lruList  *list.List
	mu       sync.Mutex // 確保線程安全
}

type entry struct {
	key        int
	value      int
	expiration time.Time // 過(guò)期時(shí)間
}

// NewCache 創(chuàng)建新的緩存
func NewCache(capacity int) *Cache {
	return &Cache{
		capacity: capacity,
		cache:    make(map[int]*list.Element),
		lruList:  list.New(),
	}
}

// Get 從緩存中獲取值
func (c *Cache) Get(key int) (int, bool) {
	c.mu.Lock()
	defer c.mu.Unlock()

	if elem, found := c.cache[key]; found {
		// 檢查是否過(guò)期
		if elem.Value.(entry).expiration.After(time.Now()) {
			// 移動(dòng)到鏈表頭部 (最近使用)
			c.lruList.MoveToFront(elem)
			return elem.Value.(entry).value, true
		}
		// 如果過(guò)期，刪除緩存項(xiàng)
		c.removeElement(elem)
	}
	return 0, false
}

// Put 將值放入緩存
func (c *Cache) Put(key int, value int, ttl time.Duration) {
	c.mu.Lock()
	defer c.mu.Unlock()

	if elem, found := c.cache[key]; found {
		// 更新現(xiàn)有值
		elem.Value = entry{key, value, time.Now().Add(ttl)}
		c.lruList.MoveToFront(elem)
	} else {
		// 添加新條目
		if c.lruList.Len() == c.capacity {
			// 刪除最舊的條目
			oldest := c.lruList.Back()
			if oldest != nil {
				c.removeElement(oldest)
			}
		}
		newElem := c.lruList.PushFront(entry{key, value, time.Now().Add(ttl)})
		c.cache[key] = newElem
	}
}

// removeElement 從緩存中刪除元素
func (c *Cache) removeElement(elem *list.Element) {
	c.lruList.Remove(elem)
	delete(c.cache, elem.Value.(entry).key)
}

// 清理過(guò)期項(xiàng)
func (c *Cache) CleanUp() {
	c.mu.Lock()
	defer c.mu.Unlock()

	for e := c.lruList.Back(); e != nil; {
		next := e.Prev()
		if e.Value.(entry).expiration.Before(time.Now()) {
			c.removeElement(e)
		}
		e = next
	}
}

func main() {
	cache := NewCache(2)
	cache.Put(1, 1, 5*time.Second) // 設(shè)置過(guò)期時(shí)間為5秒
	cache.Put(2, 2, 5*time.Second)

	fmt.Println(cache.Get(1)) // 輸出: 1 true
	time.Sleep(6 * time.Second)

	// 過(guò)期后的訪問(wèn)
	fmt.Println(cache.Get(1)) // 輸出: 0 false
	cache.CleanUp()           // 進(jìn)行清理
}

該代碼中使用LRU算法，通過(guò)將最新的緩存移動(dòng)到鏈表頭部（最近使用）來(lái)實(shí)現(xiàn)這個(gè)算法。但也有一些問(wèn)題，CleanUp 方法需要手動(dòng)調(diào)用去清理過(guò)期緩存，并沒(méi)有定期自動(dòng)清理的機(jī)制。這就意味著使用者可能需要頻繁調(diào)用 CleanUp，否則過(guò)期項(xiàng)可能會(huì)在緩存中停留較長(zhǎng)時(shí)間。代碼中也加了鎖，可能還會(huì)存在并發(fā)訪問(wèn)時(shí)的數(shù)據(jù)一致性和性能問(wèn)題等等。

所以，這種自己實(shí)現(xiàn)的demo不建議放在生產(chǎn)環(huán)境中使用，可能會(huì)存在一些小問(wèn)題。比如，之前我部門寫的一個(gè)本地緩存類庫(kù)，就存在一個(gè)大的bug：本地緩存的內(nèi)存空間不能釋放，導(dǎo)致內(nèi)存一直蹭蹭地往上漲。隔好幾天內(nèi)存就飆到90%，然后我們臨時(shí)處理方法是：隔好幾天就去重啟一次腳本...

2、所以呢，我們還是建議去使用開源的類庫(kù)，至少有許多前輩幫我們踩過(guò)坑了，這里推薦幾個(gè)star數(shù)比較高的：

（1）go-cache：一個(gè)簡(jiǎn)單的內(nèi)存緩存庫(kù)，支持過(guò)期和自動(dòng)清理，適合簡(jiǎn)單的緩存key-value需求。（本人項(xiàng)目中使用比較多，方便簡(jiǎn)單，推薦）

（2）bigcache：高性能的內(nèi)存緩存庫(kù)，適用于大量數(shù)據(jù)的緩存，其設(shè)計(jì)旨在減少垃圾回收的壓力。

（3）groupcache：Google 開發(fā)的一個(gè)緩存庫(kù)，支持分布式緩存和單機(jī)緩存。適用于需要高并發(fā)和高可用性的場(chǎng)景。

以上，就是個(gè)人使用本地緩存的一些經(jīng)驗(yàn)了。不得不說(shuō)，這玩意用著是真香，物美價(jià)廉，能扛能打。唯一美中不足的就是本地緩存不太好實(shí)時(shí)去更新，當(dāng)然這個(gè)上面也給出了幾個(gè)解決方案。

到此這篇關(guān)于golang高并發(fā)之本地緩存詳解的文章就介紹到這了,更多相關(guān)go本地緩存內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！