dy_server/service/main/redis/redis.go

package redis

import (
	"app/service/main/message/pb"
	"context"
	"encoding/json"
	"fmt"
	"github.com/redis/go-redis/v9"
)

type RankInfo struct {
	OpenId string
	Score  int32
}

var client *redis.Client

func init() {
	url := "redis://:adhd@123@101.35.201.220:6379/1?protocol=3"
	opts, err := redis.ParseURL(url)
	if err != nil {
		panic(err)
	}
	client = redis.NewClient(opts)
}

func AddScore(appId string, scoreMap map[string]int32) error {
	ctx := context.Background()
	pip := client.Pipeline()
	for openId, score := range scoreMap {
		key := fmt.Sprintf("Score_%s", appId)
		pip.ZIncrBy(ctx, key, float64(score), openId)
	}
	_, err := pip.Exec(ctx)
	return err
}

func GetRank(appId string, topCount int32) ([]*RankInfo, error) {
	ctx := context.Background()
	key := fmt.Sprintf("Score_%s", appId)
	cmd := client.ZRevRangeWithScores(ctx, key, 0, int64(topCount))
	result, err := cmd.Result()
	if err != nil {
		return nil, err
	}
	rankList := make([]*RankInfo, 0)
	for _, info := range result {
		rankList = append(rankList, &RankInfo{
			OpenId: info.Member.(string),
			Score:  int32(info.Score),
		})
	}
	return rankList, nil
}

func SetAudience(data *pb.Audience) {
	ctx := context.Background()
	key := fmt.Sprintf("UserData_%s", data.OpenId)
	exist, err := client.Exists(ctx, key).Result()
	if err != nil {
		return
	}
	if exist == 1 {
		return
	}

	jsonData, err := json.Marshal(data)
	if err != nil {
		return
	}
	client.Set(ctx, key, string(jsonData), -1)
}

func GetAudience(openId string) *pb.Audience {
	ctx := context.Background()
	key := fmt.Sprintf("UserData_%s", openId)
	result, err := client.Get(ctx, key).Result()
	if err != nil {
		return nil
	}
	data := &pb.Audience{}
	err = json.Unmarshal([]byte(result), data)
	if err != nil {
		return nil
	}
	return data
}

func Subscribe(appId string, roomId string, pushFunc func(string), pushCloseChan chan struct{}) {
	ps := client.Subscribe(context.Background(), PublishKey(appId, roomId))
	go func() {
		for {
			msg, err := ps.ReceiveMessage(context.Background())
			if err != nil {
				return
			}
			pushFunc(msg.Payload)
		}
	}()
	go func() {
		<-pushCloseChan
		_ = ps.Unsubscribe(context.Background(), PublishKey(appId, roomId))
	}()
}

func Publish(appId string, roomId string, data *pb.NotifyAudienceAction) {
	jsonData, err := json.Marshal(data)
	if err != nil {
		return
	}
	client.Publish(context.Background(), PublishKey(appId, roomId), string(jsonData))
}

func PublishKey(appId string, roomId string) string {
	return fmt.Sprintf("DataPush_%s_%s", appId, roomId)
}