用Go写一个缓存工具

前言

在项目发开过程中，一直有用到本地缓存和分布式本地缓存，最近从Java转到Go，也需要在Go里面重新实现下这个缓存工具。

本地缓存：堆内缓存，访问速度快，系统重启后缓存清除。

分布式本地缓存：分布式本地缓存本质上还是本地缓存，只有在分布式环境下，本地缓存数据不同步，但是有时候为了快速访问做了这样的一个缓存工具，就是在某个节点缓存变化之后，立即通知其他节点清除其他节点的缓存key，重新从数据库同步缓存数据。

代码实现

本地缓存

在java里本地缓存的实现是通过Guava Cache来实现的，查看Guava Cache可以参照之前写的一篇文章：Guava常用工具# Cache本地缓存_guava 缓存-CSDN博客，那么在Go里面这里我们选择常用的BigCache来替代Guava Cache，有关BigCache的知识可以线下去了解，实现也比较简单。

1. 定义缓存接口模型

   package cache
   
   //这里使用泛型，可以放入任何类型的数据
   type Cache[T any] interface {
   
       //添加缓存
   	Put(key string, t T)
   
       //获取缓存
   	Get(key string) T
   
       //删除缓存
   	EvictKey(key string)
   
       //批量删除缓存
   	EvictKeys(keys []string)
   
       //获取所有缓存key
   	GetAllKeys() []string
   }
   

   缓存接口模型是缓存工具对外使用的基础接口，具体实现的缓存类型需要实现这些接口。

2. 创建本地缓存

   package cache
   
   import (
   	"bytes"
   	"encoding/gob"
   	"reflect"
   
   	"git.qingteng.cn/ms-public/qtmf/logx"
   	"github.com/allegro/bigcache"
   )
   
   //localcache里有个Bigcache类型，缓存的操作也是对Bigcache的操作
   type LocalCache[T any] struct {
   	Cache *bigcache.BigCache
   }
   
   
   //加入缓存，注意这里有个序列化的问题，如果我们出入的对象不能进行序列化，这里会出现报错
   func (lc *LocalCache[T]) Put(key string, t T) {
   	var buf bytes.Buffer
   	enc := gob.NewEncoder(&buf)
   	err := enc.Encode(t)
   	if err != nil {
   		logx.WithoutContext().Error("local cache encode cache value error", key, err)
   		return
   	}
   	data := buf.Bytes()
   	lc.Cache.Set(key, data)
   }
   
   //获取缓存
   func (lc *LocalCache[T]) Get(key string) T {
   	var value T
   	data, err := lc.Cache.Get(key)
   	if err != nil {
   		logx.WithoutContext().Error("local cache get cache value error", key, err)
   		return reflect.New(reflect.TypeOf(value)).Elem().Interface().(T)
   	}
   
   	if data == nil {
   		return reflect.New(reflect.TypeOf(value)).Elem().Interface().(T)
   	}
   
   	buf := bytes.NewBuffer(data)
   	dec := gob.NewDecoder(buf)
   	err = dec.Decode(&value)
   	if err != nil {
   		logx.WithoutContext().Error("local cache decode cache value error", key, err)
   		return value
   	}
   	return value
   }
   
   //删除缓存
   func (lc *LocalCache[T]) EvictKey(key string) {
   	lc.Cache.Delete(key)
   }
   
   
   //批量删除缓存
   func (lc *LocalCache[T]) EvictKeys(keys []string) {
   	for _, key := range keys {
   		lc.Cache.Delete(key)
   	}
   }
   
   //获取所有缓存key
   func (lc *LocalCache[T]) GetAllKeys() []string {
   	var keys []string
   
   	// 遍历 bigcache
   	iterator := lc.Cache.Iterator()
   	for iterator.SetNext() {
   		current, _ := iterator.Value()
   		keys = append(keys, current.Key())
   	}
   	return keys
   }

到这里本地缓存的工具类就已经实现完了，我们只需要创建一个LocaclCache结构体即可开始使用，使用方式后续会介绍。?

分布式本地缓存

分布式本地缓存还是一个本地缓存，只不过多了一个分布式环境节点通知的步骤，这个通知我们采用的是redis的pud/sub机制，所以可以想象下分布式本地缓存的一个构造，肯定会有个bigcache属性用来做实际的存储，另外还需要有一个消息广播器来往其他节点广播消息。

1. 定义分布式缓存结构体

   package cache
   
   type HeapDistributedCache[T any] struct {
   	LocalCache  *LocalCache[T]    //本地缓存
   	Config      *DistributedCacheConfig  //缓存的配置，广播的时候需要用到
   	Broadcaster *Broadcaster[T]  //消息广播器
   }
   
   //添加缓存并且进行消息广播
   func (hc *HeapDistributedCache[T]) Put(key string, t T) {
   	hc.LocalCache.Put(key, t)
   	if hc.Config.Broadcast {
   		hc.Broadcaster.broadcastEvict(hc.Config.Channel, []string{key})
   	}
   }
   
   //获取缓存
   func (hc *HeapDistributedCache[T]) Get(key string) T {
   	return hc.LocalCache.Get(key)
   }
   
   //删除缓存并且进行消息广播
   func (hc *HeapDistributedCache[T]) EvictKey(key string) {
   	hc.LocalCache.EvictKey(key)
   	if hc.Config.Broadcast {
   		hc.Broadcaster.broadcastEvict(hc.Config.Channel, []string{key})
   	}
   }
   
   //删除缓存并且进行消息广播
   func (hc *HeapDistributedCache[T]) EvictKeys(keys []string) {
   	hc.LocalCache.EvictKeys(keys)
   	if hc.Config.Broadcast {
   		hc.Broadcaster.broadcastEvict(hc.Config.Channel, keys)
   	}
   }
   
   //获取所有缓存key
   func (hc *HeapDistributedCache[T]) GetAllKeys() []string {
   	return hc.LocalCache.GetAllKeys()
   }

2. 消息广播器
   消息广播器代码示例：

   package cache
   
   import (
   	"context"
   
   	"git.qingteng.cn/ms-app-ids/service-ids/internal/common/util"
   
   	"git.qingteng.cn/ms-app-ids/service-ids/internal/broadcast"
   	"git.qingteng.cn/ms-app-ids/service-ids/internal/infra/logw"
   	"github.com/go-redis/redis/v8"
   )
   
   type Broadcaster[T any] struct {
   	RedisClient redis.UniversalClient  //redis客户端
   	Node        *broadcast.Node   //节点信息
   }
   
   //注册消息通道
   func (b *Broadcaster[T]) doSubscribe(channel string, localCache *LocalCache[T]) {
   	pubsub := b.RedisClient.Subscribe(context.Background(), channel)
   	ch := pubsub.Channel()
       //注册消息通道成功后会有一个channel返回，后续这个通道有消息都会发过来，这里用select- 
       //channel的方式进行监听
   	go func() {
   		for {
   			select {
   			case msg := <-ch:
   				{
   					b.handleMessage(msg, b.Node.Id, localCache)
   				}
   			}
   		}
   	}()
   }
   
   //收到消息之后进行的处理
   func (b *Broadcaster[T]) handleMessage(msg *redis.Message, currentNodeId string, localCache *LocalCache[T]) {
   	logger := logw.WithoutContext()
   	var broadCastCacheMsg BroadcastCacheMessage
   	err := util.Unmarshal([]byte(msg.Payload), &broadCastCacheMsg)
   	if err != nil {
   		logger.Error("receiver cache broadcast cache msg failed", broadCastCacheMsg.Action, broadCastCacheMsg.NodeId)
   	}
   
       //如果是当前节点发出的消息不需要处理，因为当前节点的缓存已经更新过
   	if currentNodeId == broadCastCacheMsg.NodeId {
   		logger.Info("receiver save node msg")
   		return
   	}
   
       //不是当前节点的，说明缓存有更新，那就直接把缓存清除，后续会重新加载最新数据
   	if broadCastCacheMsg.Action == ActionEvict {
   		localCache.EvictKeys(broadCastCacheMsg.Keys)
   	}
   }
   
   //广播缓存key清除消息
   func (b *Broadcaster[T]) broadcastEvict(channel string, keys []string) {
   	message := &BroadcastCacheMessage{
   		NodeId: b.Node.Id,
   		Action: ActionEvict,
   		Keys:   keys,
   	}
   	b.publishCacheMsg(channel, message)
   }
   
   //广播缓存清空消息
   func (b *Broadcaster[T]) broadcastClear(channel string) {
   	message := &BroadcastCacheMessage{
   		NodeId: b.Node.Id,
   		Action: ActionEvict,
   	}
   	b.publishCacheMsg(channel, message)
   }
   
   //最终发布消息
   func (b *Broadcaster[T]) publishCacheMsg(channel string, message *BroadcastCacheMessage) {
   	logger := logw.WithoutContext()
   	sendMsgBytes, err := util.Marshal(message)
   	if err != nil {
   		logger.Error("send cache broadcast msg failed", "channel", channel, err)
   	}
   
   	err = b.RedisClient.Publish(context.Background(), channel, string(sendMsgBytes)).Err()
   	if err != nil {
   		logger.Error("send cache broadcast msg error", "channel", channel, "msg", message)
   		return
   	}
   }

   广播的时候用到了本地的一个节点，所以我们需要定义一个当前节点，可以想到这个节点是个单例，有个唯一的ID。
   节点示例：

   package broadcast
   
   import (
   	"sync"
   
   	uuidX "github.com/google/uuid"
   )
   
   var (
   	nodeInstance *Node
   	mutex        sync.Mutex
   )
   
   type Message struct {
   	NodeId  string
   	MsgType messageType
   	Body    string
   }
   
   type Node struct {
   	Id string
   }
   
   // GetNodeInstance 获取node单例对象
   func GetNodeInstance() *Node {
   	mutex.Lock()
   	defer mutex.Unlock()
   	if nodeInstance != nil {
   		return nodeInstance
   	}
   	nodeInstance = &Node{
   		Id: uuidX.New().String(),
   	}
   	return nodeInstance
   }

   缓存配置结构体定义：

   package cache
   
   import "time"
   
   const (
   	ActionEvict = "evict"
   )
   
   type BroadcastCacheMessage struct {
   	NodeId string
   	Action string
   	Keys   []string
   }
   
   type DistributedCacheConfig struct {
   	Ttl       time.Duration
   	Channel   string
   	Broadcast bool
   }

   
   
   消息广播器的实现就是一个发布订阅的机制，Redis的发布订阅是基于channel来的，就跟topic一样，每个缓存可以指定channel，然后监听对应的channel即可。

缓存构造器

我们创建好了两种缓存的底层实现，那么还需要定义一个缓存构造器，方便使用。

缓存构造器cache_builder示例：

package cache

import (
	"time"

	"git.qingteng.cn/ms-app-ids/service-ids/internal/broadcast"
	"git.qingteng.cn/ms-app-ids/service-ids/internal/infra/logw"
	"git.qingteng.cn/ms-public/qtmf/providers/redisx"
	"github.com/allegro/bigcache"
)

var (
	DefaultSecond       = 300
	DefaultCleanCWindow = 3
)

// ConfigLocalCache ttl传值需要跟上跟上单位，直接填入数字默认的单位是纳秒
func ConfigLocalCache[T any](ttl time.Duration) (*LocalCache[T], error) {
	if ttl <= 0 {
		ttl = time.Duration(DefaultSecond) * time.Second
	}
	config := bigcache.DefaultConfig(ttl)
	config.CleanWindow = time.Duration(DefaultCleanCWindow) * time.Second
	bigCache, err := bigcache.NewBigCache(config)
	if err != nil {
		logw.WithoutContext().Error("create local cache error")
		return nil, err
	}

	return &LocalCache[T]{
		Cache: bigCache,
	}, nil
}

func DefaultLocalCache[T any]() (*LocalCache[T], error) {
	config := bigcache.DefaultConfig(time.Duration(DefaultSecond) * time.Second)
	config.CleanWindow = time.Duration(DefaultCleanCWindow) * time.Second
	bigCache, err := bigcache.NewBigCache(config)
	if err != nil {
		logw.WithoutContext().Error("create local cache error")
		return nil, err
	}

	return &LocalCache[T]{
		Cache: bigCache,
	}, nil
}

//构建分布式本地缓存
func ConfigHeapDistributedCache[T any](config *DistributedCacheConfig) (*HeapDistributedCache[T], error) {
	if config.Ttl <= 0 {
		config.Ttl = time.Duration(DefaultSecond) * time.Second
	}
	localCache, err := ConfigLocalCache[T](config.Ttl)
	if err != nil {
		return nil, err
	}
	redisClient, err := redisx.Client()
	if err != nil {
		return nil, err
	}
	broadcaster := &Broadcaster[T]{
		RedisClient: redisClient,
		Node:        broadcast.GetNodeInstance(),
	}
	if config.Broadcast {
		broadcaster.doSubscribe(config.Channel, localCache)
	}
	return &HeapDistributedCache[T]{
		LocalCache:  localCache,
		Config:      config,
		Broadcaster: broadcaster,
	}, nil
}

缓存测试
?

func TestLoaclCache(t *testing.T) {
	intcache, _ := cache.ConfigLocalCache[int](10 * time.Second)
	intcache.Put("test", 111)
	intValue := intcache.Get("test")
	println(intValue)

	stringCache, _ := cache.DefaultLocalCache[string]()
	stringCache.Put("test2", "hello world")
	stringValue := stringCache.Get("test2")
	println(stringValue)

	personCache, _ := cache.DefaultLocalCache[*Person]()
	person := &Person{
		Name: "张三",
		Age:  20,
	}
	personCache.Put("person", person)
	personCache.Put("person2", &Person{})
	presult := personCache.Get("person")
	println(presult.Name, presult.Age)
	fmt.Println(personCache.GetAllKeys())
}


localCache测试输出：
111
hello world
张三 20
[person person2]


func TestHeapDistributeCache(t *testing.T) {
	config := &cache.DistributedCacheConfig{
		Ttl:       5,
		Channel:   "test_heap_distribute",
		Broadcast: true,
	}
	stringcache, _ := cache.ConfigHeapDistributedCache[string](config)
	stringcache.Put("test_heap", "test_heap")
	println(stringcache.Get("test_heap"))

	pconfig := &cache.DistributedCacheConfig{
		Ttl:       5,
		Channel:   "test_person_distribute",
		Broadcast: true,
	}
	personCache, _ := cache.ConfigHeapDistributedCache[*Person](pconfig)
	person := &Person{
		Name: "张三",
		Age:  20,
	}
	personCache.Put("test_heap_person", person)
	presult := personCache.Get("test_heap_person")
	println(presult.Name, presult.Age)

	time.Sleep(time.Second * 5)
}

分布式缓存测试输出：
test_heap
张三 20