go-kratos 学习笔记(7) 服务发现服务间通信grpc调用

服务发现

Registry 接口分为两个，Registrar 为实例注册和反注册，Discovery 为服务实例列表获取

创建一个 Discoverer

服务间的通信使用的grpc，放到data层，实现的是从uses服务调用orders服务

app/users/internal/data.go 加入 NewDiscovery和 NewOrderServiceClient，需要把新加的2个方法加入到 ProviderSet

需要把新生成的orderClient注入到Data里面 orderClient orders.OrderClient

package dataimport ("context""github.com/go-kratos/kratos/contrib/registry/nacos/v2""github.com/go-kratos/kratos/v2/log""github.com/go-kratos/kratos/v2/middleware/recovery""github.com/go-kratos/kratos/v2/registry""github.com/go-kratos/kratos/v2/transport/grpc""github.com/google/wire""github.com/nacos-group/nacos-sdk-go/clients""github.com/nacos-group/nacos-sdk-go/common/constant""github.com/nacos-group/nacos-sdk-go/vo""gorm.io/driver/mysql""gorm.io/gorm""time""xgs_kratos/gen/config/users""xgs_kratos/gen/orders"
)// ProviderSet is data providers.
var ProviderSet = wire.NewSet(NewData, NewDiscovery, CreateRegister, NewOrderServiceClient, NewUserRepo)// Data .
type Data struct {// TODO wrapped database clientdb          *gorm.DBlog         *log.HelperorderClient orders.OrderClient
}// NewData .func NewData(c *conf.Data, logger log.Logger, client orders.OrderClient) (*Data, func(), error) {cleanup := func() {log.NewHelper(logger).Info("closing the data resources")}db, err := gorm.Open(mysql.Open(c.Database.Source), &gorm.Config{})if err != nil {log.Fatalf("failed to connect database: %v", err)panic(err)}return &Data{db:          db,log:         log.NewHelper(logger),orderClient: client,}, cleanup, nil
}// NewDiscovery 服务发现
func NewDiscovery(conf *conf.Data) registry.Discovery {sc := []constant.ServerConfig{{IpAddr: conf.Nacos.Addr,Port:   conf.Nacos.Port,},}cc := constant.ClientConfig{NamespaceId: conf.Nacos.NamespaceId,TimeoutMs:   5000,}client, err := clients.NewNamingClient(vo.NacosClientParam{ClientConfig:  &cc,ServerConfigs: sc,},)if err != nil {panic(err)}r := nacos.New(client)return r
}// NewOrderServiceClient orders 服务客户端
func NewOrderServiceClient(r registry.Discovery) orders.OrderClient {conn, err := grpc.DialInsecure(context.Background(),grpc.WithEndpoint("discovery:///orders-xgs.grpc"),grpc.WithDiscovery(r),grpc.WithTimeout(time.Second*2),grpc.WithMiddleware(recovery.Recovery(),),)if err != nil {panic(err)}c := orders.NewOrderClient(conn)return c
}

在users下执行 wire

以ListUser方法为例子进行调用

app/users/internal/user.go

package dataimport ("context""fmt""github.com/go-kratos/kratos/v2/log""xgs_kratos/app/users/internal/biz""xgs_kratos/app/users/internal/data/dal""xgs_kratos/gen/orders""xgs_kratos/gen/users"
)//data 层处理数据的存储和读取type userRepo struct {data *Datalog  *log.Helper
}// NewUserRepo . r registry.Discovery,
func NewUserRepo(data *Data, logger log.Logger) biz.UserRepo {return &userRepo{data: data,log:  log.NewHelper(logger),}
}// CreateUser 创建用户
func (r *userRepo) CreateUser(ctx context.Context, req *users.CreateUserRequest) (*users.CreateUserReply, error) {user := dal.UserMo{Age:   req.Age,Name:  req.Name,Email: req.Email,}result := r.data.db.Create(&user)if result.Error != nil {return nil, result.Error}return &users.CreateUserReply{Id: user.Id,}, nil
}func (r *userRepo) ListUser(ctx context.Context, req *users.ListUserRequest) ([]*users.UserData, error) {//获取order服务的clientclient := r.data.orderClientorder, err := client.CreateOrder(ctx, &orders.CreateOrderRequest{OrderNo: 1,})if err != nil {return nil, err}fmt.Println(order)var results []dal.UserMores := r.data.db.Find(&results)if res.Error != nil {return nil, res.Error}var userDatas []*users.UserDatafor _, result := range results {userDatas = append(userDatas, &users.UserData{Id:    result.Id,Name:  result.Name,Age:   result.Age,Email: result.Email,})}return userDatas, nil
}

从新生成一下代码 buf generate

如果服务是分开部署的，需要拿到对方的存根 proto 执行 kratos proto client xxx.proto生成client

分别启动users和orders 服务是都是注册成功的，由于注册的是http和grpc所有后面拼接的有http和grpc，调用的时候需要拼接上