利用go-kit组件进行服务注册与发现和健康检查的操作

 type Reg struct { Host string Port int Client consul.Client } func MakeReg (host string , port int) (*Reg , error) { reg := api.DefaultConfig() reg.Address = host + ":" + strconv.Itoa(port) apiclient , err = api.NewClient(reg) if err != nil { return nil , err } client := consul.NewClient(apiclient) return &Reg{Host : host , Port : port ,Client : client} , nil } func (r Reg) Resiter (servicename , tag , host , seviceid ,checkinter ,healthcheckhttp ,deregisterafter string , port int) bool { congig := api.AgentServiceRegistration{ Port : port , Address : host , Name := servicename, ID := serviceid, Ckeck : &api.AgentServiceCheck{ Interval : checkinter, HTTP : "http://" + host + ":" + healthcheckhttp , DeregisterCriticalServiceAfter : deregisterafter, } } if err := r.Client.Register(&config) ; err != nil { fmt.Println(err) return false } return true } func (r Reg) Deregister (serviceid string) bool { dreg := api.AgentServiceRegistration{ID : serviceid} if err != r.Client.Deregister(&config) ; err != nil { fmt.Println(err) return false } return true } func (r Reg) discover (servicename , tag string ,passingonly bool) ( []*api.ServiceEntry ,error ) { if entries ,_ ,err := r.Client.Service(servicename , tag , passingonly , nil) ;err != nil { return nil ,err }else{ return entries , nil } } 

 type Balancer interface { Endpoint() (endpoint.Endpoint, error) }

 var ( //etcd服务地址 etcdServer = "127.0.0.1:2379" //服务的信息目录 prefix     = "/services/book/" //当前启动服务实例的地址 instance   = "127.0.0.1:50052" //服务实例注册的路径 key        = prefix + instance //服务实例注册的val value      = instance ctx        = context.Background() //服务监听地址 serviceAddress = ":50052" ) //etcd的连接参数 options := etcdv3.ClientOptions{ DialTimeout: time.Second * 3, DialKeepAlive: time.Second * 3, } //创建etcd连接 client, err := etcdv3.NewClient(ctx, []string{etcdServer}, options) if err != nil { panic(err) } // 创建注册器 registrar := etcdv3.NewRegistrar(client, etcdv3.Service{ Key:   key, Value: value, }, log.NewNopLogger()) // 注册器启动注册 registrar.Register()

 package main import ( "grpc-test/pb" "context" grpc_transport "github.com/go-kit/kit/transport/grpc" "github.com/go-kit/kit/endpoint" "google.golang.org/grpc" "net" "github.com/go-kit/kit/sd/etcdv3" "github.com/go-kit/kit/log" "time" ) type BookServer struct { bookListHandler  grpc_transport.Handler bookInfoHandler  grpc_transport.Handler } //通过grpc调用GetBookInfo时,GetBookInfo只做数据透传, 调用BookServer中对应Handler.ServeGRPC转交给go-kit处理 func (s *BookServer) GetBookInfo(ctx context.Context, in *book.BookInfoParams) (*book.BookInfo, error) { _, rsp, err := s.bookInfoHandler.ServeGRPC(ctx, in) if err != nil { return nil, err } return rsp.(*book.BookInfo),err } //通过grpc调用GetBookList时,GetBookList只做数据透传, 调用BookServer中对应Handler.ServeGRPC转交给go-kit处理 func (s *BookServer) GetBookList(ctx context.Context, in *book.BookListParams) (*book.BookList, error) { _, rsp, err := s.bookListHandler.ServeGRPC(ctx, in) if err != nil { return nil, err } return rsp.(*book.BookList),err } //创建bookList的EndPoint func makeGetBookListEndpoint() endpoint.Endpoint { return func(ctx context.Context, request interface{}) (interface{}, error) { //请求列表时返回 书籍列表 bl := new(book.BookList) bl.BookList = append(bl.BookList, &book.BookInfo{BookId:1,BookName:"21天精通php"}) bl.BookList = append(bl.BookList, &book.BookInfo{BookId:2,BookName:"21天精通java"}) return bl,nil } } //创建bookInfo的EndPoint func makeGetBookInfoEndpoint() endpoint.Endpoint { return func(ctx context.Context, request interface{}) (interface{}, error) { //请求详情时返回 书籍信息 req := request.(*book.BookInfoParams) b := new(book.BookInfo) b.BookId = req.BookId b.BookName = "21天精通php" return b,nil } } func decodeRequest(_ context.Context, req interface{}) (interface{}, error) { return req, nil } func encodeResponse(_ context.Context, rsp interface{}) (interface{}, error) { return rsp, nil } func main() { var ( //etcd服务地址 etcdServer = "127.0.0.1:2379" //服务的信息目录 prefix     = "/services/book/" //当前启动服务实例的地址 instance   = "127.0.0.1:50052" //服务实例注册的路径 key        = prefix + instance //服务实例注册的val value      = instance ctx        = context.Background() //服务监听地址 serviceAddress = ":50052" ) //etcd的连接参数 options := etcdv3.ClientOptions{ DialTimeout: time.Second * 3, DialKeepAlive: time.Second * 3, } //创建etcd连接 client, err := etcdv3.NewClient(ctx, []string{etcdServer}, options) if err != nil { panic(err) } // 创建注册器 registrar := etcdv3.NewRegistrar(client, etcdv3.Service{ Key:   key, Value: value, }, log.NewNopLogger()) // 注册器启动注册 registrar.Register() bookServer := new(BookServer) bookListHandler := grpc_transport.NewServer( makeGetBookListEndpoint(), decodeRequest, encodeResponse, ) bookServer.bookListHandler = bookListHandler bookInfoHandler := grpc_transport.NewServer( makeGetBookInfoEndpoint(), decodeRequest, encodeResponse, ) bookServer.bookInfoHandler = bookInfoHandler ls, _ := net.Listen("tcp", serviceAddress) gs := grpc.NewServer(grpc.UnaryInterceptor(grpc_transport.Interceptor)) book.RegisterBookServiceServer(gs, bookServer) gs.Serve(ls) }

 package main import ( "context" "github.com/go-kit/kit/sd/etcdv3" "time" "github.com/go-kit/kit/sd" "github.com/go-kit/kit/log" "github.com/go-kit/kit/endpoint" "io" "github.com/go-kit/kit/sd/lb" "grpc-test/pb" "fmt" "google.golang.org/grpc" ) func main() { var ( //注册中心地址 etcdServer = "127.0.0.1:2379" //监听的服务前缀 prefix     = "/services/book/" ctx        = context.Background() ) options := etcdv3.ClientOptions{ DialTimeout: time.Second * 3, DialKeepAlive: time.Second * 3, } //连接注册中心 client, err := etcdv3.NewClient(ctx, []string{etcdServer}, options) if err != nil { panic(err) } logger := log.NewNopLogger() //创建实例管理器, 此管理器会Watch监听etc中prefix的目录变化更新缓存的服务实例数据 instancer, err := etcdv3.NewInstancer(client, prefix, logger) if err != nil { panic(err) } //创建端点管理器， 此管理器根据Factory和监听的到实例创建endPoint并订阅instancer的变化动态更新Factory创建的endPoint endpointer := sd.NewEndpointer(instancer, reqFactory, logger) //创建负载均衡器 balancer := lb.NewRoundRobin(endpointer) /** 我们可以通过负载均衡器直接获取请求的endPoint，发起请求 reqEndPoint,_ := balancer.Endpoint() */ /** 也可以通过retry定义尝试次数进行请求 */ reqEndPoint := lb.Retry(3, 3*time.Second, balancer) //现在我们可以通过 endPoint 发起请求了 req := struct{}{} if _, err = reqEndPoint(ctx, req); err != nil { panic(err) } } //通过传入的 实例地址  创建对应的请求endPoint func reqFactory(instanceAddr string) (endpoint.Endpoint, io.Closer, error) { return func(ctx context.Context, request interface{}) (interface{}, error) { fmt.Println("请求服务: ", instanceAddr) conn, err := grpc.Dial(instanceAddr, grpc.WithInsecure()) if err != nil { fmt.Println(err) panic("connect error") } defer conn.Close() bookClient := book.NewBookServiceClient(conn) bi,_:=bookClient.GetBookInfo(context.Background(),&book.BookInfoParams{BookId:1}) fmt.Println("获取书籍详情") fmt.Println("bookId: 1", " => ", "bookName:", bi.BookName) bl,_ := bookClient.GetBookList(context.Background(), &book.BookListParams{Page:1, Limit:10}) fmt.Println("获取书籍列表") for _,b := range bl.BookList { fmt.Println("bookId:", b.BookId, " => ", "bookName:", b.BookName) } return nil,nil },nil,nil } 

 请求服务： 127.0.0.1:50052 获取书籍详情 bookId: 1  =>  bookName: 21天精通php 获取书籍列表 bookId: 1  =>  bookName: 21天精通php bookId: 2  =>  bookName: 21天精通java 

 instance   = "127.0.0.1:50052" serviceAddress = ":50052"

 req := struct{}{} for i := 1; i <= 8; i++ { if _, err = reqEndPoint(ctx, req); err != nil { panic(err) } }

 请求服务:  127.0.0.1:50051 获取书籍详情 bookId: 1  =>  bookName: 21天精通php 获取书籍列表 bookId: 1  =>  bookName: 21天精通php bookId: 2  =>  bookName: 21天精通java 请求服务:  127.0.0.1:50052 获取书籍详情 bookId: 1  =>  bookName: 21天精通php 获取书籍列表 bookId: 1  =>  bookName: 21天精通php bookId: 2  =>  bookName: 21天精通java 请求服务:  127.0.0.1:50051 获取书籍详情 bookId: 1  =>  bookName: 21天精通php 获取书籍列表 bookId: 1  =>  bookName: 21天精通php bookId: 2  =>  bookName: 21天精通java 请求服务:  127.0.0.1:50052 获取书籍详情 bookId: 1  =>  bookName: 21天精通php 获取书籍列表 bookId: 1  =>  bookName: 21天精通php bookId: 2  =>  bookName: 21天精通java 请求服务:  127.0.0.1:50051 获取书籍详情 bookId: 1  =>  bookName: 21天精通php 获取书籍列表 bookId: 1  =>  bookName: 21天精通php bookId: 2  =>  bookName: 21天精通java 请求服务:  127.0.0.1:50052 获取书籍详情 bookId: 1  =>  bookName: 21天精通php 获取书籍列表 bookId: 1  =>  bookName: 21天精通php bookId: 2  =>  bookName: 21天精通java 请求服务:  127.0.0.1:50051 获取书籍详情 bookId: 1  =>  bookName: 21天精通php 获取书籍列表 bookId: 1  =>  bookName: 21天精通php bookId: 2  =>  bookName: 21天精通java 请求服务:  127.0.0.1:50052 获取书籍详情 bookId: 1  =>  bookName: 21天精通php 获取书籍列表 bookId: 1  =>  bookName: 21天精通php bookId: 2  =>  bookName: 21天精通java Process finished with exit code 0