命令记不住

缓存服务代码详解

1.go-cache

初始化缓存：

main.go

// 数据缓存
var DBCache *cache.Cache = cache.New(30*time.Minute, 10*time.Minute)

实现

blogdataimpl.go

// ArticleQueryWithId
func (s *BlogDataServer) ArticleQueryWithId(ctx context.Context, in *pb.ArticleQueryWithIdRequest) (*pb.ArticleQueryWithIdResponse, error) {
	now := time.Now()
	fmt.Printf("ArticleQueryWithId: %s\n", now.Format(TIME_FORMAT))
	cachekey := fmt.Sprintf("article_%d", in.Id)
	if x, found := DBCache.Get(cachekey); found {
		fmt.Printf("hit: %s\n", cachekey)
		fmt.Printf("duration: %v\n", time.Now().Sub(now))
		return x.(*pb.ArticleQueryWithIdResponse), nil
	}

	article := models.QueryArticleWithId(int(in.Id))

	result := &pb.ArticleQueryWithIdResponse{
		Article: models.ArticleToPb(article),
	}
	fmt.Printf("set: %s\n", cachekey)
	DBCache.Set(cachekey, result, cache.DefaultExpiration)
	fmt.Printf("duration: %v\n", time.Now().Sub(now))
	return result, nil
}

运行代码，观察server日志

这时发现，修改文章后，缓存没有及时刷新

2.redis

命令查询

https://redis.io/commands

代码接入

// ArticleQueryWithId
func (s *BlogDataServer) ArticleQueryWithId(ctx context.Context, in *pb.ArticleQueryWithIdRequest) (*pb.ArticleQueryWithIdResponse, error) {
	now := time.Now()
	fmt.Printf("ArticleQueryWithId: %s\n", now.Format(TIME_FORMAT))
	// Establish a connection to the Redis server listening on port
	// 6379 of the local machine. 6379 is the default port, so unless
	// you've already changed the Redis configuration file this should
	// work.
	conn, err := redis.Dial("tcp", "localhost:6379")
	if err != nil {
		log.Fatal(err)
	}
	// Importantly, use defer to ensure the connection is always
	// properly closed before exiting the main() function.
	defer conn.Close()

	cachekey := fmt.Sprintf("article_%d", in.Id)
	// if x, found := DBCache.Get(cachekey); found {
	// 	fmt.Printf("hit: %s\n", cachekey)
	// 	fmt.Printf("duration: %v\n", time.Now().Sub(now))
	// 	return x.(*pb.ArticleQueryWithIdResponse), nil
	// }

	article := models.QueryArticleWithId(int(in.Id))

	result := &pb.ArticleQueryWithIdResponse{
		Article: models.ArticleToPb(article),
	}
	fmt.Printf("set: %s\n", cachekey)
	// DBCache.Set(cachekey, result, cache.DefaultExpiration)
	jsondata, _ := json.Marshal(result)
	conn.Do("SET", cachekey, jsondata, "EX", 3600)
	fmt.Printf("duration: %v\n", time.Now().Sub(now))
	return result, nil
}

3.更新数据

找到消息消费者

esutils.go

// 更新一个ES文档
func SearchArticleUpdate(id int) {
	fmt.Printf("SearchArticleUpdate: %d", id)
	ctx := context.Background()
	client, err := elastic.NewSimpleClient(elastic.SetURL("http://127.0.0.1:9200/"))
	if err != nil {
		// Handle error
		fmt.Println(err.Error())
		return
	}
	defer client.Stop()
	item := QueryArticleWithId(id)
	fmt.Printf("SearchArticleUpdate: %d", item.Id)
	se_index := "blogweb_article"
	se_type := "doc"
	se_id := fmt.Sprintf("%d", item.Id)
	e := item.ToSearchEntity()

	put1, err := client.Index().
		Index(se_index).
		Type(se_type).
		Id(se_id).
		BodyJson(e).
		Do(ctx)
	if err != nil {
		// Handle error
		fmt.Println(err.Error())
	}
	fmt.Printf("Indexed %s to index %s, type %s\n", put1.Id, put1.Index, put1.Type)

	// 删除索引
	conn, err := redis.Dial("tcp", "localhost:6379")
	if err != nil {
		log.Fatal(err)
	}
	// Importantly, use defer to ensure the connection is always
	// properly closed before exiting the main() function.
	defer conn.Close()
	cachekey := fmt.Sprintf("article_%d", id)
	fmt.Printf("delete %s\n", cachekey)
	_, err = conn.Do("DEL", cachekey)
	if err != nil {
		log.Fatal(err)
	}
}

4.连接池的用法

连接池 or actor model

连接池的用法

package main

import (
	"errors"

	"github.com/gomodule/redigo/redis"
)

// Declare a pool variable to hold the pool of Redis connections.
var pool *redis.Pool

func YourMethod(id string) (*Album, error) {
	// Use the connection pool's Get() method to fetch a single Redis
	// connection from the pool.
	conn := pool.Get()

	// Importantly, use defer and the connection's Close() method to
	// ensure that the connection is always returned to the pool before
	// FindAlbum() exits.
	defer conn.Close()
	...
}

func main() {
	// Initialize a connection pool and assign it to the pool global
	// variable.
	pool = &redis.Pool{
		MaxIdle:     10,
		IdleTimeout: 240 * time.Second,
		Dial: func() (redis.Conn, error) {
			return redis.Dial("tcp", "localhost:6379")
		},
	}

	...
}

actor model

func Start() {
	once.Do(func() {
		//index := int32(1)
		l4g.Debug("msgstore::Start()")
		var client redis.Client
		l4g.Debug("addr_redis: %s", configsvr.Fetch("addr_redis"))
		client.Addr = configsvr.Fetch("addr_redis")
		client.Db = 1
		defer func() {
			close(msgStoreChans.chGetUnsent)
			close(msgStoreChans.chAddMessage)
			close(msgStoreChans.chListAllData)
			close(msgStoreChans.chSendAck)
		}()
		for {
			select {
			case value := <-msgStoreChans.chGetUnsent:
				key := "C_" + value.ClientId
				val, err := client.Lrange(key, 0, -1)
				if err != nil {
					l4g.Error(err)
				} else {
					l4g.Debug("lrange value: %v", val)
				}
				result := []*TMessage{}
                ...
				value.ReplyChan <- result
			case value := <-msgStoreChans.chAddMessage:
				l4g.Debug("", value.clientId)
				client.Setex("M_"+value.msgId, value.time, []byte(value.msgBody))
				client.Rpush("C_"+value.clientId, []byte(value.msgId))
			case replyChannel := <-msgStoreChans.chListAllData:
				l4g.Debug("msgStoreChans.chListAllData")
				keys, _ := client.Keys("M_*")
				result := []*TKeyValue{}
                ...
				replyChannel <- result
			case value := <-msgStoreChans.chSendAck:
				client.Del("M_" + value.MessageId)
				client.Lrem1("C_"+value.ClientId, 0, []byte(value.MessageId))
			}
		}
	})
}