package rpc import ( "context" "fmt" "git.hilo.cn/hilo-common/mylogrus" "git.hilo.cn/hilo-common/protocol/userCenter" "git.hilo.cn/hilo-common/resource/config" "git.hilo.cn/hilo-common/resource/consul" "git.hilo.cn/hilo-common/sdk/dingding" consulapi "github.com/hashicorp/consul/api" "google.golang.org/grpc" "google.golang.org/grpc/keepalive" "google.golang.org/grpc/resolver" "time" ) const ( Port = 50060 ) var userClient userCenter.UserClient 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 } var ( defaultUserCenterAddr = "127.0.0.1:50040" // userCenter default addr userCenterAddr = defaultUserCenterAddr userCenterConsulName = "userCenter" ) // grpc服务发现 type Builder struct { addrs map[string][]string cc resolver.ClientConn } func (b *Builder) Scheme() string { return "uc" // userCenter } type Resolver struct { } func (r Resolver) ResolveNow(options resolver.ResolveNowOptions) {} func (r Resolver) Close() {} func (b *Builder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) { r := &Resolver{} paths := b.addrs[target.URL.Path] addrs := make([]resolver.Address, len(paths)) for i, s := range paths { addrs[i] = resolver.Address{Addr: s} } cc.UpdateState(resolver.State{Addresses: addrs}) b.cc = cc return r, nil } func (b *Builder) UpdateState(addrs []string) { as := make([]resolver.Address, len(addrs)) for i, s := range addrs { as[i] = resolver.Address{Addr: s} } b.cc.UpdateState(resolver.State{Addresses: as}) } func init() { // 本地环境下不需要userCenter if config.AppIsLocal() { mylogrus.MyLog.Infoln("userCenter is not required in local env.") return } client, err := consulapi.NewClient(consulapi.DefaultConfig()) //非默认情况下需要设置实际的参数 if err != nil || client == nil { mylogrus.MyLog.Errorf("consul client fail:%v-%v", err, client) } // 服务发现 bd := &Builder{addrs: map[string][]string{"/api": {userCenterAddr}}} if client != nil { cataLog := client.Catalog() if cataLog != nil { services, _, err := cataLog.Service("userCenter", "", nil) if err != nil { mylogrus.MyLog.Errorf("cataLog.Service:err%v", err) } if len(services) == 0 { mylogrus.MyLog.Errorf("userCenter not found.") } var addrs []string for _, s := range services { addrs = append(addrs, fmt.Sprintf("%s:%d", s.ServiceAddress, s.ServicePort)) } if len(addrs) > 0 { bd = &Builder{addrs: map[string][]string{"/api": addrs}} userCenterAddr = "uc:///api" } mylogrus.MyLog.Infof("userCenterAddr:%v,addr:%v", userCenterAddr, addrs) } } mylogrus.MyLog.Infof("connect userCenterAddr:%v", userCenterAddr) resolver.Register(bd) go func() { address := consulapi.DefaultConfig().Address // 用consul api的default config if err := consul.RegisterWatcher("services", nil, address, func(serviceStatus map[string]map[string][]string) { if statusAddrs, ok := serviceStatus[userCenterConsulName]; ok { healthAddrs, _ := statusAddrs[consulapi.HealthPassing] l := len(healthAddrs) if l > 0 { mylogrus.MyLog.Infof("consul service update state:%v-%v", userCenterConsulName, healthAddrs) bd.UpdateState(healthAddrs) // 更新新的注册名 } else { mylogrus.MyLog.Warnf("consul service update local state:%v-%v", userCenterConsulName, defaultUserCenterAddr) bd.UpdateState([]string{defaultUserCenterAddr}) // 都没有健康的,使用默认本地回环的 } for status := range statusAddrs { if status == consulapi.HealthPassing { continue } mylogrus.MyLog.Warnf("consul service wrong state:%v-%v-%v", userCenterConsulName, status, statusAddrs[status]) } } }); err != nil { mylogrus.MyLog.Errorf("启动 consul 的watch监控失败") } }() //userCenterAddr := services[0].Address + ":" + strconv.Itoa(services[0].ServicePort) //mylogrus.MyLog.Printf("Choose userCenter %s, %s, weights: %v\n", services[0].ID, userCenterAddr, services[0].ServiceWeights) // Set up a connection to the userCenter. conn, err := grpc.Dial(userCenterAddr, grpc.WithInsecure(), grpc.WithBlock(), grpc.WithKeepaliveParams(kacp), grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy":"%s"}`, "round_robin"))) if err != nil { msg := fmt.Sprintf("mode:%v,连接userCenter失败:%v", config.GetMode(), err) dingding.SendDingRobot(dingding.ROBOTWEBHOOK, msg, true) mylogrus.MyLog.Fatalf(msg) } //defer conn.Close() userClient = userCenter.NewUserClient(conn) if userClient == nil { msg := fmt.Sprintf("mode:%v,userClient null", config.GetMode()) dingding.SendDingRobot(dingding.ROBOTWEBHOOK, msg, true) mylogrus.MyLog.Fatalln(msg) } } func multicast(uids []uint64, msgType uint32, data []byte) ([]uint64, error) { ctx, cancel := context.WithTimeout(context.Background(), time.Second*3) defer cancel() rsp, err := userClient.Multicast(ctx, &userCenter.MulticastMessage{ Uids: uids, MsgType: msgType, PayLoad: data, }) if err != nil { mylogrus.MyLog.Errorf("Multicast message failed %s", err.Error()) } if rsp != nil { mylogrus.MyLog.Infof("Multicast message res:%+v", rsp) return rsp.FailedUids, err } else { return []uint64{}, err } } //广播 func broadcast(msgType uint32, data []byte) ([]uint64, error) { ctx, cancel := context.WithTimeout(context.Background(), time.Second*3) 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 } }