go-common/utils/redis/redis_go_cluster_adapter.go

package redis

import (
	"fmt"
	redisc "github.com/chasex/redis-go-cluster"
	"github.com/spf13/viper"
	"time"
)

// redis-go-cluster适配器
type GoRedisClusterAdapter struct {
	cluster *redisc.Cluster
}

// 返回redis-go-cluster适配器新实例
func NewGoRedisClusterAdapter(nodes []string) (IRedisAdapter, error) {
	maxActive := viper.GetInt("redis.max_active")
	idleTimeout := viper.GetDuration("redis.timeout")

	cluster, err := redisc.NewCluster(&redisc.Options{
		StartNodes:  nodes,
		ConnTimeout: 5 * time.Second,
		KeepAlive:   maxActive,
		AliveTime:   idleTimeout,
	})
	if nil != err {
		return nil, err
	}
	return &GoRedisClusterAdapter{
		cluster: cluster,
	}, nil
}

// 关闭Redis连接
func (a *GoRedisClusterAdapter) Close() error {
	a.cluster.Close()
	return nil
}

// 执行Redis命令
func (a *GoRedisClusterAdapter) Do(commandName string, args ...interface{}) (interface{}, error) {
	return a.cluster.Do(commandName, args...)
}

// 返回命令管道操作对象
func (a *GoRedisClusterAdapter) Pipeline() IRedisPipeline {
	return &GoRedisClusterPipeline{
		cluster: a.cluster,
	}
}

// Redigo命令管道操作对象
type GoRedisClusterPipeline struct {
	cluster  *redisc.Cluster
	commands []*redisCmd
}

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

// 执行管道中的命令
func (p *GoRedisClusterPipeline) Execute() ([]interface{}, error) {
	if nil == p.commands || len(p.commands) == 0 {
		return nil, fmt.Errorf("no commands in pipeline")
	}

	batch := p.cluster.NewBatch()

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