go-common/utils/redis/redigo_adapter.go

package redis

import (
	"fmt"
	"github.com/gomodule/redigo/redis"
	"github.com/spf13/viper"
	"strings"
	"time"
)

// redigo适配器
type RedigoAdapter struct {
	pool *redis.Pool
}

// 返回redigo适配器新实例
func NewRedigoAdapter(addr string) (IRedisAdapter, error) {
	opts := []redis.DialOption{redis.DialConnectTimeout(5 * time.Second)}
	password := viper.GetString("redis.password")
	if len(strings.TrimSpace(password)) > 0 {
		opts = append(opts, redis.DialPassword(strings.TrimSpace(password)))
	}
	pool, err := createPool(addr, opts...)
	if nil != err {
		return nil, err
	}
	return &RedigoAdapter{
		pool: pool,
	}, nil
}

// 创建redis连接池
func createPool(addr string, opts ...redis.DialOption) (*redis.Pool, error) {
	maxIdle := viper.GetInt("redis.max_idle")
	maxActive := viper.GetInt("redis.max_active")
	idleTimeout := viper.GetDuration("redis.timeout")

	return &redis.Pool{
		MaxIdle:     maxIdle,
		MaxActive:   maxActive,
		IdleTimeout: idleTimeout,
		Dial: func() (redis.Conn, error) {
			if conn, err := redis.Dial("tcp", addr, opts...); nil != err {
				return nil, err
			} else {
				return conn, nil
			}
		},
		TestOnBorrow: func(c redis.Conn, t time.Time) error {
			_, err := c.Do("PING")
			return err
		},
	}, nil
}

// 关闭Redis连接
func (a *RedigoAdapter) Close() error {
	return a.pool.Close()
}

// 执行Redis命令
func (a *RedigoAdapter) Do(commandName string, args ...interface{}) (interface{}, error) {
	conn := a.pool.Get()
	defer func(conn redis.Conn) {
		_ = conn.Close()
	}(conn)

	if nil == conn {
		return nil, ErrRedisConnNil
	} else {
		return conn.Do(commandName, args...)
	}
}

// 返回命令管道操作对象
func (a *RedigoAdapter) Pipeline() IRedisPipeline {
	return &RedigoPipeline{
		conn: a.pool.Get(),
	}
}

// Redigo命令管道操作对象
type RedigoPipeline struct {
	conn     redis.Conn
	commands []*redisCmd
}

// 向管道中添加命令
func (p *RedigoPipeline) Send(commandName string, args ...interface{}) IRedisPipeline {
	p.commands = append(p.commands, &redisCmd{commandName: commandName, args: args})
	return p
}

// 执行管道中的命令
func (p *RedigoPipeline) Execute() ([]interface{}, error) {
	defer func(conn redis.Conn) {
		_ = conn.Close()
	}(p.conn)

	if nil == p.commands || len(p.commands) == 0 {
		return nil, fmt.Errorf("no commands in pipeline")
	}

	var err error
	for i := 0; i < len(p.commands); i++ {
		cmd := p.commands[i]
		err = p.conn.Send(cmd.commandName, cmd.args...)
		if nil != err {
			return nil, err
		}
	}
	err = p.conn.Flush()
	if nil != err {
		return nil, err
	}

	var replies []interface{}
	for i := 0; i < len(p.commands); i++ {
		reply, err := p.conn.Receive()
		if nil != err {
			return nil, err
		}
		replies = append(replies, reply)
	}
	return replies, nil
}