GO XORM+Redis 自己整合

心已赠人 2023-07-24 08:03 19阅读 0赞

1. 先写一个redis的工具类。

package xormrediscache
    
    import (
    	"bytes"
    	"encoding/gob"
    	"fmt"
    	"hash/crc32"
    	"os"
    
    	"strings"
    
    	"github.com/gomodule/redigo/redis"
    	"github.com/xormplus/core"
    
    	// "log"
    	"reflect"
    	"time"
    	"unsafe"
    )
    
    const (
    	DEFAULT_EXPIRATION = time.Duration(0)
    	FOREVER_EXPIRATION = time.Duration(-1)
    
    	LOGGING_PREFIX = "[redis_cacher]"
    )
    
    // Wraps the Redis client to meet the Cache interface.
    type RedisCacher struct {
    	pool              *redis.Pool
    	defaultExpiration time.Duration
    
    	Logger core.ILogger
    }
    
    // New a Redis Cacher, host as IP endpoint, i.e., localhost:6379, provide empty string or nil if Redis server doesn't
    // require AUTH command, defaultExpiration sets the expire duration for a key to live. Until redigo supports
    // sharding/clustering, only one host will be in hostList
    //
    //     engine.SetDefaultCacher(xormrediscache.NewRedisCacher("localhost:6379", "", xormrediscache.DEFAULT_EXPIRATION, engine.Logger))
    //
    // or set MapCacher
    //
    //     engine.MapCacher(&user, xormrediscache.NewRedisCacher("localhost:6379", "", xormrediscache.DEFAULT_EXPIRATION, engine.Logger))
    //
    var (
    	REDIS_ENABLE   = os.Getenv("REDIS_ENABLE")
    	REDIS_HOST     = os.Getenv("REDIS_HOST")
    	REDIS_PORT     = os.Getenv("REDIS_PORT")
    	REDIS_PASSWORD = os.Getenv("REDIS_PASSWORD")
    	REDIS_DATABASE = os.Getenv("REDIS_DATABASE")
    	// REDIS_EXPIRATION = os.Getenv("REDIS_EXPIRATION")
    )
    
    // 是否使用reds
    func RedisEnable() bool {
    	if strings.EqualFold(REDIS_ENABLE, "false") {
    		return false
    	}
    	if strings.EqualFold(REDIS_ENABLE, "true") {
    		return true
    	}
    	return false
    }
    func NewRedisCacher(defaultExpiration time.Duration, logger core.ILogger) *RedisCacher {
    	// 环境变量过期时间
    	// if REDIS_EXPIRATION != "" && REDIS_EXPIRATION != "-1" {
    	// 	if expiration, err := strconv.ParseInt(REDIS_EXPIRATION, 10, 64); nil == err {
    	// 		// 默认使用环境变量的配置 单位分钟
    	// 		defaultExpiration = time.Duration(expiration) * time.Minute
    	// 	}
    	// }
    	// 初始化
    	if REDIS_HOST == "" {
    		REDIS_HOST = "127.0.0.1"
    	}
    	if REDIS_PORT == "" {
    		REDIS_PORT = "6379"
    	}
    	if REDIS_DATABASE == "" {
    		REDIS_DATABASE = "3"
    	}
    	str := REDIS_HOST + ":" + REDIS_PORT
    	logger.Debugf("Starting connect redis:%s", str)
    	var pool = &redis.Pool{
    		MaxIdle:     16,
    		MaxActive:   32,
    		IdleTimeout: 240 * time.Second,
    		Dial: func() (redis.Conn, error) {
    			// the redis protocol should probably be made sett-able
    			c, err := redis.Dial("tcp", str)
    			if err != nil {
    				logger.Errorf("Connect redis HOST error: %s", err.Error())
    				return nil, err
    			} else {
    				logger.Debugf("Connect redis HOST successfully.")
    			}
    			if len(REDIS_PASSWORD) > 0 {
    				if _, err := c.Do("AUTH", REDIS_PASSWORD); err != nil {
    					c.Close()
    					logger.Errorf("Connect redis AUTH error: %s", err.Error())
    					return nil, err
    				} else {
    					logger.Debugf("Connect redis AUTH successfully.")
    				}
    
    			} else {
    				// check with PING
    				if _, err := c.Do("PING"); err != nil {
    					c.Close()
    					return nil, err
    				}
    			}
    			if _, err := c.Do("SELECT", REDIS_DATABASE); err != nil {
    				c.Close()
    				logger.Errorf("SELECT database %s error:%s.", REDIS_DATABASE, err.Error())
    				return nil, err
    			} else {
    				logger.Debugf("SELECT database %s successfully.", REDIS_DATABASE)
    			}
    			return c, err
    		},
    		// custom connection test method
    		TestOnBorrow: func(c redis.Conn, t time.Time) error {
    			if _, err := c.Do("PING"); err != nil {
    				return err
    			}
    			return nil
    		},
    	}
    	return MakeRedisCacher(pool, defaultExpiration, logger)
    }
    
    // MakeRedisCacher build a cacher based on redis.Pool
    func MakeRedisCacher(pool *redis.Pool, defaultExpiration time.Duration, logger core.ILogger) *RedisCacher {
    	return &RedisCacher{pool: pool, defaultExpiration: defaultExpiration, Logger: logger}
    }
    
    func exists(conn redis.Conn, key string) bool {
    	existed, _ := redis.Bool(conn.Do("EXISTS", key))
    	return existed
    }
    
    func (c *RedisCacher) logErrf(format string, contents ...interface{}) {
    	if c.Logger != nil {
    		c.Logger.Errorf(fmt.Sprintf("%s %s", LOGGING_PREFIX, format), contents...)
    	}
    }
    
    func (c *RedisCacher) logDebugf(format string, contents ...interface{}) {
    	if c.Logger != nil {
    		c.Logger.Debugf(fmt.Sprintf("%s %s", LOGGING_PREFIX, format), contents...)
    	}
    }
    
    func (c *RedisCacher) getBeanKey(tableName string, id string) string {
    	return fmt.Sprintf("xorm:bean:%s:%s", tableName, id)
    }
    
    func (c *RedisCacher) getSqlKey(tableName string, sql string) string {
    	// hash sql to minimize key length
    	crc := crc32.ChecksumIEEE([]byte(sql))
    	return fmt.Sprintf("xorm:sql:%s:%d", tableName, crc)
    }
    
    // Delete current database
    func (c *RedisCacher) Flushdb() error {
    	if !RedisEnable() {
    		return nil
    	}
    	conn := c.pool.Get()
    	defer conn.Close()
    	_, err := conn.Do("flushdb")
    	return err
    
    }
    
    // Delete by product and is
    func (c *RedisCacher) FlushdbProductId(tableName string, productId, id uint64) error {
    	if !RedisEnable() {
    		return nil
    	}
    	var key string
    	if productId == 0 {
    
    	} else {
    		if id == 0 {
    			// 删除产品对应的所有数据
    			key = fmt.Sprintf("product_id:%d:id:", productId) + "*"
    		} else {
    			// 删除所有的id
    			key = fmt.Sprintf("product_id:%d:id:%d:", productId, id) + "*"
    		}
    	}
    
    	return c.delObjects(c.getBeanKey(strings.ToLower(tableName), key))
    }
    
    // Delete all db
    func (c *RedisCacher) Flush() error {
    	if !RedisEnable() {
    		return nil
    	}
    	conn := c.pool.Get()
    	defer conn.Close()
    	_, err := conn.Do("flushall")
    	return err
    
    }
    func (c *RedisCacher) getObject(key string) interface{} {
    
    	conn := c.pool.Get()
    	defer conn.Close()
    	raw, err := conn.Do("GET", key)
    	if raw == nil {
    		return nil
    	}
    	item, err := redis.Bytes(raw, err)
    	if err != nil {
    		c.logErrf("redis.Bytes failed: %s", err)
    		return nil
    	}
    
    	value, err := c.deserialize(item)
    
    	return value
    }
    
    func (c *RedisCacher) GetIds(tableName, sql string) interface{} {
    	if !RedisEnable() {
    		return nil
    	}
    	sqlKey := c.getSqlKey(tableName, sql)
    	c.logDebugf(" GetIds|tableName:%s|sql:%s|key:%s", tableName, sql, sqlKey)
    	return c.getObject(sqlKey)
    }
    
    // 默认过期资源
    func (c *RedisCacher) GetBean(tableName string, id string) interface{} {
    	if !RedisEnable() {
    		return nil
    	}
    	beanKey := c.getBeanKey(tableName, id)
    	c.logDebugf("[xorm/redis_cacher] GetBean|tableName:%s|id:%s|key:%s", tableName, id, beanKey)
    	return c.getObject(beanKey)
    }
    
    // 默认过期资源
    func (c *RedisCacher) putObject(key string, value interface{}) {
    	c.invoke(c.pool.Get().Do, key, value, c.defaultExpiration)
    }
    
    // 自定义过期资源
    func (c *RedisCacher) putObjectExpiration(key string, value interface{}, defaultExpiration time.Duration) {
    	c.invoke(c.pool.Get().Do, key, value, defaultExpiration)
    }
    
    // 自定义过期资源
    func (c *RedisCacher) PutIdsExpiration(tableName, sql string, ids interface{}, defaultExpiration time.Duration) {
    	if !RedisEnable() {
    		return
    	}
    	sqlKey := c.getSqlKey(tableName, sql)
    	c.logDebugf("PutIds|tableName:%s|sql:%s|key:%s|obj:%s|type:%v", tableName, sql, sqlKey, ids, reflect.TypeOf(ids))
    	c.putObjectExpiration(sqlKey, ids, defaultExpiration)
    }
    
    // 自定义过期资源
    func (c *RedisCacher) PutBeanExpiration(tableName string, id string, obj interface{}, defaultExpiration time.Duration) {
    	if !RedisEnable() {
    		return
    	}
    	beanKey := c.getBeanKey(tableName, id)
    	c.logDebugf("PutBean|tableName:%s|id:%s|key:%s|type:%v", tableName, id, beanKey, reflect.TypeOf(obj))
    	c.putObjectExpiration(beanKey, obj, defaultExpiration)
    }
    
    func (c *RedisCacher) PutIds(tableName, sql string, ids interface{}) {
    	if !RedisEnable() {
    		return
    	}
    	sqlKey := c.getSqlKey(tableName, sql)
    	c.logDebugf("PutIds|tableName:%s|sql:%s|key:%s|obj:%s|type:%v", tableName, sql, sqlKey, ids, reflect.TypeOf(ids))
    	c.putObject(sqlKey, ids)
    }
    
    func (c *RedisCacher) PutBean(tableName string, id string, obj interface{}) {
    	if !RedisEnable() {
    		return
    	}
    	beanKey := c.getBeanKey(tableName, id)
    	c.logDebugf("PutBean|tableName:%s|id:%s|key:%s|type:%v", tableName, id, beanKey, reflect.TypeOf(obj))
    	c.putObject(beanKey, obj)
    }
    
    func (c *RedisCacher) delObject(key string) error {
    	c.logDebugf("delObject key:[%s]", key)
    
    	conn := c.pool.Get()
    	defer conn.Close()
    	if !exists(conn, key) {
    		c.logErrf("delObject key:[%s] err: %v", key, core.ErrCacheMiss)
    		return core.ErrCacheMiss
    	}
    	_, err := conn.Do("DEL", key)
    	return err
    }
    
    func (c *RedisCacher) delObjects(key string) error {
    
    	c.logDebugf("delObjects key:[%s]", key)
    
    	conn := c.pool.Get()
    	defer conn.Close()
    
    	keys, err := conn.Do("KEYS", key)
    	c.logDebugf("delObjects keys: %v", keys)
    
    	if err == nil {
    		for _, key := range keys.([]interface{}) {
    			conn.Do("DEL", key)
    		}
    	}
    	return err
    }
    
    func (c *RedisCacher) DelIds(tableName, sql string) {
    	if !RedisEnable() {
    		return
    	}
    	c.delObject(c.getSqlKey(tableName, sql))
    }
    
    func (c *RedisCacher) DelBean(tableName string, id string) {
    	if !RedisEnable() {
    		return
    	}
    	c.delObject(c.getBeanKey(tableName, id))
    }
    func (c *RedisCacher) DelBeans(tableName string, id string) {
    	if !RedisEnable() {
    		return
    	}
    	c.delObjects(c.getBeanKey(tableName, id))
    }
    func (c *RedisCacher) ClearIds(tableName string) {
    	if !RedisEnable() {
    		return
    	}
    	c.delObjects(fmt.Sprintf("xorm:sql:%s:*", tableName))
    }
    
    func (c *RedisCacher) ClearBeans(tableName string) {
    	if !RedisEnable() {
    		return
    	}
    	c.delObjects(c.getBeanKey(tableName, "*"))
    }
    
    func (c *RedisCacher) invoke(f func(string, ...interface{}) (interface{}, error),
    	key string, value interface{}, expires time.Duration) error {
    
    	switch expires {
    	case DEFAULT_EXPIRATION:
    		expires = c.defaultExpiration
    	case FOREVER_EXPIRATION:
    		expires = time.Duration(0)
    	}
    
    	b, err := c.serialize(value)
    	if err != nil {
    		return err
    	}
    	conn := c.pool.Get()
    	defer conn.Close()
    	if expires > 0 {
    		_, err := f("SETEX", key, int32(expires/time.Second), b)
    		return err
    	} else {
    		_, err := f("SET", key, b)
    		return err
    	}
    }
    
    func (c *RedisCacher) serialize(value interface{}) ([]byte, error) {
    
    	err := c.registerGobConcreteType(value)
    	if err != nil {
    		return nil, err
    	}
    
    	if reflect.TypeOf(value).Kind() == reflect.Struct {
    		return nil, fmt.Errorf("serialize func only take pointer of a struct")
    	}
    
    	var b bytes.Buffer
    	encoder := gob.NewEncoder(&b)
    
    	c.logDebugf("serialize type:%v", reflect.TypeOf(value))
    	err = encoder.Encode(&value)
    	if err != nil {
    		c.logErrf("gob encoding '%s' failed: %s|value:%v", value, err, value)
    		return nil, err
    	}
    	return b.Bytes(), nil
    }
    
    func (c *RedisCacher) deserialize(byt []byte) (ptr interface{}, err error) {
    	b := bytes.NewBuffer(byt)
    	decoder := gob.NewDecoder(b)
    
    	var p interface{}
    	err = decoder.Decode(&p)
    	if err != nil {
    		c.logErrf("decode failed: %v", err)
    		return
    	}
    
    	v := reflect.ValueOf(p)
    	c.logDebugf("deserialize type:%v", v.Type())
    	if v.Kind() == reflect.Struct {
    
    		var pp interface{} = &p
    		datas := reflect.ValueOf(pp).Elem().InterfaceData()
    
    		sp := reflect.NewAt(v.Type(),
    			unsafe.Pointer(datas[1])).Interface()
    		ptr = sp
    		vv := reflect.ValueOf(ptr)
    		c.logDebugf("deserialize convert ptr type:%v | CanAddr:%t", vv.Type(), vv.CanAddr())
    	} else {
    		ptr = p
    	}
    	return
    }
    
    func (c *RedisCacher) registerGobConcreteType(value interface{}) error {
    
    	t := reflect.TypeOf(value)
    
    	c.logDebugf("registerGobConcreteType:%v", t)
    
    	switch t.Kind() {
    	case reflect.Ptr:
    		v := reflect.ValueOf(value)
    		i := v.Elem().Interface()
    		gob.Register(&i)
    	case reflect.Struct, reflect.Map, reflect.Slice:
    		gob.Register(value)
    	case reflect.String, reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64, reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64, reflect.Bool, reflect.Float32, reflect.Float64, reflect.Complex64, reflect.Complex128:
    		// do nothing since already registered known type
    	default:
    		return fmt.Errorf("unhandled type: %v", t)
    	}
    	return nil
    }
    
    func (c *RedisCacher) GetPool() (*redis.Pool, error) {
    	return c.pool, nil
    }
    
    func (c *RedisCacher) SetPool(pool *redis.Pool) {
    	c.pool = pool
    }

2. 写一个xorm 统一入口的增删改查的方法

// 删除或者更新使用的sql
    func DeleteOrUpdateCache(tableName, primaryKey, sqlstr string) (int, error) {
    	//DB().Logger().Debugf(sqlstr)
    	if ret, err := DB().Exec(sqlstr); nil == err {
    		if rows, _ := ret.RowsAffected(); rows > 0 {
    			// 清空缓存
    			if (int)(rows) > 0 {
    				RedisCacher.DelBeans(tableName, primaryKey)
    			}
    
    			return (int)(rows), nil
    		} else {
    			return 0, errors.New(datatype.ErrorNotFound.Message)
    		}
    	} else {
    		//数据库或SQL错误记录日志，级别为Error。
    		DB().Logger().Errorf(err.Error() + config.ToolmakerErrorLogTag)
    		return 0, errors.New(datatype.ErrorDatabase.Message)
    	}
    }
    
    // update 更新sql
    func Update(session *xorm.Session, tableName, primaryKey string, bean interface{}, condiBean ...interface{}) (int64, error) {
    	// 清空緩存
    	rec, err := session.Update(bean)
    	if nil == err && rec != 0 {
    		// 清空緩存
    		RedisCacher.DelBeans(tableName, primaryKey)
    	}
    	return rec, err
    }
    
    // 查询单跳记录的sql 有过期时间
    func GetOneCacheExpiration(tableName, primaryKey, sqlstr string, obj interface{}, defaultExpiration time.Duration) error {
    	// 查询缓存中的数据
    	redisRecord := RedisCacher.GetBean(tableName, primaryKey)
    
    	if nil == redisRecord {
    		if ret, err := DB().SQL(sqlstr).Get(obj); nil == err {
    			if ret == true {
    				// 更新到缓存中
    				if byteStr, jsonerr := json.Marshal(&obj); nil == jsonerr {
    					RedisCacher.PutBeanExpiration(tableName, primaryKey, string(byteStr), defaultExpiration)
    				}
    				return nil
    			} else {
    				return errors.New(datatype.ErrorNotFound.Message)
    			}
    		} else {
    			//数据库或SQL错误记录日志，级别为Error。
    			DB().Logger().Errorf(err.Error() + config.ToolmakerErrorLogTag)
    			return errors.New(datatype.ErrorDatabase.Message)
    		}
    	} else {
    		// 装换返回
    		json.Unmarshal([]byte(redisRecord.(string)), &obj)
    		return nil
    	}
    
    }

涉及了xorm自带的update方法和正常的sql凭借删除修改查询方法

后面所有的sql 同一个用这些封装的方法 即满足了 缓存的使用

参考:

[https://github.com/go-xorm/xorm-redis-cache][https_github.com_go-xorm_xorm-redis-cache]

[https://github.com/xormplus/xorm][https_github.com_xormplus_xorm]

[https_github.com_go-xorm_xorm-redis-cache]: https://github.com/go-xorm/xorm-redis-cache
[https_github.com_xormplus_xorm]: https://github.com/xormplus/xorm