feat 结构调整

Server/common/net/grpc/resolver/conn.go

@@ -0,0 +1,30 @@
+package resolver
+
+import (
+	"common/log"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/keepalive"
+	"time"
+)
+
+func NewGrpcConnection(target string) (*grpc.ClientConn, error) {
+	cc, err := grpc.NewClient(
+		target,
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithDefaultServiceConfig(`{"loadBalancingConfig": [{"round_robin": {}}]}`),
+		grpc.WithKeepaliveParams(
+			keepalive.ClientParameters{
+				Time:                30 * time.Second, // 保活探测包发送的时间间隔
+				Timeout:             10 * time.Second, // 保活探测包的超时时间
+				PermitWithoutStream: true,
+			},
+		),
+		//grpc.WithStatsHandler(&StatsHandler{}),
+	)
+	if err != nil {
+		log.Errorf("create grpc err: %v, target: %v", err, target)
+		return nil, err
+	}
+	return cc, nil
+}

Server/common/net/grpc/resolver/etcd_resolver.go

@@ -0,0 +1,84 @@
+package resolver
+
+import (
+	"common/db/etcd"
+	"common/discover/common"
+	"common/log"
+	"context"
+	"go.etcd.io/etcd/client/v3"
+	"google.golang.org/grpc/resolver"
+	"strings"
+)
+
+const etcdSchema = "etcd"
+
+func init() {
+	resolver.Register(&etcdBuilder{})
+}
+
+type etcdBuilder struct{}
+
+func (*etcdBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
+	serviceName := strings.TrimPrefix(target.URL.Path, "/")
+	r := &etcdResolver{
+		serviceName: serviceName,
+		cc:          cc,
+	}
+	r.start()
+	return r, nil
+}
+
+func (*etcdBuilder) Scheme() string { return etcdSchema }
+
+type etcdResolver struct {
+	serviceName string
+	cc          resolver.ClientConn
+	ctx         context.Context
+	cancel      context.CancelFunc
+}
+
+func (r *etcdResolver) start() {
+	r.ctx, r.cancel = context.WithCancel(context.Background())
+	r.update()
+	go r.watch()
+}
+
+func (r *etcdResolver) getPrefix() string {
+	return common.KeyDiscoverService + "/" + r.serviceName
+}
+
+func (r *etcdResolver) update() {
+	resp, err := etcd.GetClient().Get(r.getPrefix(), clientv3.WithPrefix())
+	if err != nil {
+		log.Errorf("etcd resolver get error: %v", err)
+		return
+	}
+
+	var addrArray []resolver.Address
+	for _, kv := range resp.Kvs {
+		addr := string(kv.Value)
+		if addr != "" {
+			addrArray = append(addrArray, resolver.Address{Addr: addr})
+		}
+	}
+
+	_ = r.cc.UpdateState(resolver.State{Addresses: addrArray})
+}
+
+func (r *etcdResolver) watch() {
+	watchCh := etcd.GetClient().Watch(r.getPrefix(), clientv3.WithPrefix())
+	for w := range watchCh {
+		if w.Err() != nil {
+			continue
+		}
+		r.update()
+	}
+}
+
+func (r *etcdResolver) ResolveNow(resolver.ResolveNowOptions) {
+	r.update()
+}
+
+func (r *etcdResolver) Close() {
+	r.cancel()
+}

Server/common/net/grpc/resolver/resolver_mgr.go

@@ -0,0 +1,35 @@
+package resolver
+
+import (
+	"google.golang.org/grpc"
+	"sync"
+)
+
+var (
+	mu   = sync.RWMutex{}
+	conn = make(map[string]*grpc.ClientConn)
+)
+
+func GetGrpcClientConn(target string) (*grpc.ClientConn, error) {
+	mu.RLock()
+	if c, ok := conn[target]; ok && c != nil {
+		mu.RUnlock()
+		return c, nil
+	}
+	mu.RUnlock()
+
+	mu.Lock()
+	defer mu.Unlock()
+
+	if c, ok := conn[target]; ok && c != nil {
+		return c, nil
+	}
+
+	newConn, err := NewGrpcConnection(target)
+	if err != nil {
+		return nil, err
+	}
+
+	conn[target] = newConn
+	return newConn, nil
+}