main.go

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/go-redis/redis/v8"
	"github.com/golang/protobuf/proto"
	"github.com/hashicorp/consul/api"
	"google.golang.org/grpc"
	"google.golang.org/grpc/keepalive"
	"google.golang.org/grpc/resolver"
	"google.golang.org/grpc/resolver/manual"
	"hilo-socketCenter/common"
	"hilo-socketCenter/common/consul"
	"hilo-socketCenter/common/dingding"
	"hilo-socketCenter/common/mylogrus"
	"hilo-socketCenter/common/redisCli"
	"hilo-socketCenter/domain/model/rpc_m"
	"hilo-socketCenter/protocol/userCenter"
	"hilo-socketCenter/protocol/userProxy"
	"time"
)

const SEND_WORKER = 1       // 消费端协程数量
const MONITOR_LENGTH = 1000 // 队列告警数量
const SocketQueueSendGift = "socket:queue:send_gift"

var userClient userCenter.UserClient

type SendGiftMsg struct {
	SendUserId uint64                      `json:"sendUserId"`
	Msg        *userProxy.GlobalGiftBanner `json:"msg"`
}

var sendGiftChan chan *SendGiftMsg

var kacp = keepalive.ClientParameters{
	Time:                10 * time.Second, // send pings every 10 seconds if there is no activity
	Timeout:             time.Second,      // wait 1 second for ping ack before considering the connection dead
	PermitWithoutStream: true,             // send pings even without active streams
}

// 初始化userCenterClient
func init() {
	client, err := api.NewClient(api.DefaultConfig()) //非默认情况下需要设置实际的参数
	if err != nil {
		mylogrus.MyLog.Fatalln(err)
	}
	cataLog := client.Catalog()
	if cataLog == nil {
		mylogrus.MyLog.Fatalln("No catalog.")
	}
	addr, err := consul.GetServices(cataLog, "userCenter")
	if err != nil {
		mylogrus.MyLog.Fatalln(err)
	}
	if len(addr) == 0 {
		mylogrus.MyLog.Fatalln("No userCenter available.")
	}

	addresses := make([]resolver.Address, len(addr))
	for i, s := range addr {
		addresses[i].Addr = s
		mylogrus.MyLog.Infof("address : %s", s)
	}

	r := manual.NewBuilderWithScheme("hilo")
	r.InitialState(resolver.State{Addresses: addresses})

	userCenterAddr := fmt.Sprintf("%s:///usercenter", r.Scheme())

	// Set up addresses connection to the userCenter.
	conn, err := grpc.Dial(userCenterAddr,
		grpc.WithInsecure(),
		grpc.WithBlock(),
		grpc.WithKeepaliveParams(kacp),
		grpc.WithResolvers(r),
		grpc.WithDefaultServiceConfig("{ \"loadBalancingPolicy\": \"round_robin\" }"))

	if err != nil {
		mylogrus.MyLog.Fatalf("did not connect: %v", err)
	}
	//defer conn.Close()
	userClient = userCenter.NewUserClient(conn)
	if userClient == nil {
		mylogrus.MyLog.Fatalln("userClient null")
	}
}

func main() {
	mylogrus.MyLog.Infof("cron sendGiftChan start")
	// 8核 n send + 4 blpop
	sendGiftChan = make(chan *SendGiftMsg, SEND_WORKER)
	for i := 0; i < 1; i++ {
		go func() {
			deal()
		}()
	}
	for i := 0; i < SEND_WORKER; i++ {
		go func() {
			send()
		}()
	}
	go check()
	select {}
}

func check() {
	tick := time.NewTicker(time.Second * 30)
	defer tick.Stop()
	for {
		select {
		case <-tick.C:
			l, err := redisCli.GetRedisCluster().LLen(context.Background(), SocketQueueSendGift).Result()
			if err != nil {
				mylogrus.MyLog.Infof("cron sendGiftChan msg error,left %v-%v", l, err)
			}
			if l > MONITOR_LENGTH {
				go func() {
					if sErr := dingding.SendDingRobot(dingding.ROBOTWEBHOOK, fmt.Sprintf("送礼横幅变化通知延迟,队列%s长度:%d", SocketQueueSendGift, l), true); sErr != nil {
						mylogrus.MyLog.Errorf("dingding msg fail:%v", sErr)
					}
				}()
				n, err := redisCli.GetRedisCluster().Del(context.Background(), SocketQueueSendGift).Result()
				mylogrus.MyLog.Infof("del sendGiftChan msg queue:%v,n:%v,err:%v", SocketQueueSendGift, n, err)
			}
			if l > 0 {
				mylogrus.MyLog.Infof("cron sendGiftChan msg,left %v", l)
			}
		}
	}
}

func deal() {
	for true {
		//不需要加锁，注意，阻塞。
		// 后进先出
		strs, err := redisCli.GetRedisCluster().BRPop(context.Background(), time.Second, SocketQueueSendGift).Result()
		if err != nil {
			if err != redis.Nil {
				mylogrus.MyLog.Errorf("cron sendGiftChan redisCli.GetRedis().BLPop err:+%v", err)
			}
		}
		if len(strs) >= 2 {
			content := strs[1]
			//mylogrus.MyLog.Infof("cron sendGiftChan content:%v", content)
			msg := new(SendGiftMsg)
			if err := json.Unmarshal([]byte(content), msg); err != nil {
				mylogrus.MyLog.Errorf("cron sendGiftChan Unmarshal err:%+v, content:%v", err, content)
			}
			sendGiftChan <- msg
		}
		time.Sleep(time.Second * 5) // 控制全服banner速率
	}
}

//var limiter = rate.NewLimiter(2000, 2000)

func send() {
	for msg := range sendGiftChan {
		//if err := limiter.Wait(context.Background()); err == nil {
		SendToUserCenter(msg.SendUserId, msg.Msg)
		//}
	}
}

// SendToUserCenter 发送群信令。入参是内部imGroupId，这里做转换
func SendToUserCenter(sendUserId uint64, msg *userProxy.GlobalGiftBanner) {
	if buffer, err := proto.Marshal(msg); err == nil {
		rspUids, err := broadcast(common.MsgTypeGlobalGiftBanner, buffer)

		//记录socket,注意闭包问题
		go func(userId uint64, msg *userProxy.GlobalGiftBanner, rspUids []uint64, err error) {
			buf, _ := json.Marshal(msg)
			rpc_m.AddRpcLog(common.MsgTypeGlobalGiftBanner, userId, string(buf[:]), rspUids, err)
		}(sendUserId, msg, rspUids, err)

		if err != nil {
			mylogrus.MyLog.Errorf("grpc GlobalGiftBanner send fail")
		} else {
			//mylogrus.MyLog.Info("grpc GlobalGiftBanner send success")
		}
	} else {
	}
}

//广播
func broadcast(msgType uint32, data []byte) ([]uint64, error) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
	defer cancel()
	rsp, err := userClient.Broadcast(ctx, &userCenter.BroadcastMessage{
		MsgType: msgType,
		PayLoad: data,
	})
	if err != nil {
		mylogrus.MyLog.Errorf("broadcast message failed %s", err.Error())
	}
	if rsp != nil {
		//mylogrus.MyLog.Infof("broadcast message res:%v", rsp)
		return rsp.FailedUids, err
	} else {
		return []uint64{}, err
	}
}