web_huangguo2_server/pub/redis/manager.go

package redis

import (
	"fmt"
	"regexp"
	"sync"
	"videoweb/pub/common/log"
	"videoweb/pub/common/srvdef"

	"github.com/go-redis/redis"
)

type SubscribeCallback func(channel RCType, message string)

var (
	redisCli *redis.Client
	opts     Options

	//订阅参数
	pubsub    *redis.PubSub
	subCBMap  map[RCType]SubscribeCallback
	cbMapLock sync.RWMutex
)

func Client() *redis.Client {
	return redisCli
}

func Disconnect() {
	if redisCli != nil {
		if pubsub != nil {
			pubsub.Close()
		}
		redisCli.Close()
		redisCli = nil
	}
}

func Init(opts Options) error {
	c := redis.NewClient(
		&redis.Options{
			Addr:     opts.Addr,
			Password: opts.Psw,
			DB:       opts.Db,
		},
	)

	if err := c.Ping().Err(); err != nil {
		log.Error(fmt.Sprintf("[error] ping redis err %+v\n", err))
		return err
	}
	redisCli = c

	pubsub = nil
	if opts.Subscribe {
		subCBMap = make(map[RCType]SubscribeCallback)
		pubsub = redisCli.Subscribe()
		//接受发布消息
		go ReceiveMessage()
	}

	log.Info("redis:Init Ok!")
	return nil
}

// OnMessage 订阅消息回调
func OnMessage(channel RCType, message string) {
	cbMapLock.RLock()
	defer cbMapLock.RUnlock()
	flag := false
	for k, v := range subCBMap {
		reg := regexp.MustCompile(string(k))
		//log.Info("test", log.Any("k", k), log.Any("channelchannel", channel))
		if b := reg.MatchString(string(channel)); b {
			v(channel, message)
			flag = true
		}
	}
	if !flag {
		log.Warn("(redis:OnMessage) callback doesn't exist!", log.Any("channel", channel))
	}
}

// ReceiveMessage 接受发布消息
func ReceiveMessage() {
	for {
		msg, err := pubsub.Receive()
		if err != nil {
			fmt.Println("test1", err)
			return
		}

		switch msg := msg.(type) {
		case *redis.Subscription:
			// Ignore.
			log.Info("(redis:ReceiveMessage) subscribe success!", log.Any("channel", msg.Channel))
		case *redis.Pong:
			// Ignore.
			fmt.Println("Pong")
		case *redis.Message:
			channel := msg.Channel
			message := msg.Payload
			OnMessage(RCType(channel), message)
		default:
			log.Error("(redis:ReceiveMessage) msg!", log.Any("msg", msg))
			return
		}
	}
}

// Subscribe 订阅一个消息
func Subscribe(channel RCType, cb SubscribeCallback) *srvdef.SrvErr {
	if pubsub == nil {
		log.Warn("(redis:Subscribe) subscribe is not enabled！")
		return srvdef.New(srvdef.ErrRdsOperError, "", "")
	}
	err := pubsub.Subscribe(string(channel))
	if err != nil {
		log.Warn("(redis:Subscribe) Subscribe failed！", log.Any("err", err))
		return srvdef.New(srvdef.ErrRdsOperError, "", "")
	}
	cbMapLock.Lock()
	defer cbMapLock.Unlock()
	subCBMap[channel] = cb
	return nil
}

// UnSubscribe 取消订阅一个消息
func UnSubscribe(channel RCType, cb SubscribeCallback) *srvdef.SrvErr {
	if pubsub == nil {
		log.Warn("(redis:UnSubscribe) subscribe is not enabled！")
		return srvdef.New(srvdef.ErrRdsOperError, "", "")
	}
	err := pubsub.Unsubscribe(string(channel))
	if err != nil {
		log.Warn("(redis:UnSubscribe) UnSubscribe failed！", log.Any("err", err))
		return srvdef.New(srvdef.ErrRdsOperError, "", "")
	}
	cbMapLock.Lock()
	defer cbMapLock.Unlock()
	delete(subCBMap, channel)
	return nil
}

// PSubscribe 模式匹配消息
func PSubscribe(channel RCType, cb SubscribeCallback) *srvdef.SrvErr {
	if pubsub == nil {
		log.Warn("(redis:PSubscribe) subscribe is not enabled！")
		return srvdef.New(srvdef.ErrRdsOperError, "", "")
	}
	err := pubsub.PSubscribe(string(channel))
	if err != nil {
		log.Warn("(redis:PSubscribe) PSubscribe failed！", log.Any("err", err))
		return srvdef.New(srvdef.ErrRdsOperError, "", "")
	}
	cbMapLock.Lock()
	defer cbMapLock.Unlock()
	subCBMap[channel] = cb
	return nil
}

// PUnsubscribe 模式取消匹配消息
func PUnsubscribe(channel RCType, cb SubscribeCallback) *srvdef.SrvErr {
	if pubsub == nil {
		log.Warn("(redis:PUnsubscribe) subscribe is not enabled！")
		return srvdef.New(srvdef.ErrRdsOperError, "", "")
	}
	err := pubsub.PUnsubscribe(string(channel))
	if err != nil {
		log.Warn("(redis:PUnsubscribe) PUnsubscribe failed！", log.Any("err", err))
		return srvdef.New(srvdef.ErrRdsOperError, "", "")
	}
	cbMapLock.Lock()
	defer cbMapLock.Unlock()
	delete(subCBMap, channel)
	return nil
}