上篇介紹瞭如何使用etcd
實現服務發現,本篇將基於etcd的服務發現前提下,介紹如何實現gRPC客戶端負載均衡。git
gRPC官方文檔提供了關於gRPC負載均衡方案Load Balancing in gRPC,此方案是爲gRPC設計的,下面咱們對此進行分析。github
gRPC中的負載平衡是以每次調用爲基礎,而不是以每一個鏈接爲基礎。換句話說,即便全部的請求都來自一個客戶端,咱們仍但願它們在全部的服務器上實現負載平衡。golang
集中式
(Proxy Model)在服務消費者和服務提供者之間有一個獨立的負載均衡(LB),一般是專門的硬件設備如 F5,或者基於軟件如 LVS,HAproxy等實現。LB上有全部服務的地址映射表,一般由運維配置註冊,當服務消費方調用某個目標服務時,它向LB發起請求,由LB以某種策略,好比輪詢(Round-Robin)作負載均衡後將請求轉發到目標服務。LB通常具有健康檢查能力,能自動摘除不健康的服務實例。segmentfault
該方案主要問題:服務消費方、提供方之間增長了一級,有必定性能開銷,請求量大時,效率較低。緩存
可能有讀者會認爲集中式負載均衡存在這樣的問題,一旦負載均衡服務掛掉,那整個系統將不能使用。
解決方案:能夠對負載均衡服務進行DNS負載均衡,經過對一個域名設置多個IP地址,每次DNS解析時輪詢返回負載均衡服務地址,從而實現簡單的DNS負載均衡。服務器
客戶端負載
(Balancing-aware Client)針對第一個方案的不足,此方案將LB的功能集成到服務消費方進程裏,也被稱爲軟負載或者客戶端負載方案。服務提供方啓動時,首先將服務地址註冊到服務註冊表,同時按期報心跳到服務註冊表以代表服務的存活狀態,至關於健康檢查,服務消費方要訪問某個服務時,它經過內置的LB組件向服務註冊表查詢,同時緩存並按期刷新目標服務地址列表,而後以某種負載均衡策略選擇一個目標服務地址,最後向目標服務發起請求。LB和服務發現能力被分散到每個服務消費者的進程內部,同時服務消費方和服務提供方之間是直接調用,沒有額外開銷,性能比較好。mvc
該方案主要問題:要用多種語言、多個版本的客戶端編寫和維護負載均衡策略,使客戶端的代碼大大複雜化。app
獨立LB服務
(External Load Balancing Service)該方案是針對第二種方案的不足而提出的一種折中方案,原理和第二種方案基本相似。負載均衡
不一樣之處是將LB和服務發現功能從進程內移出來,變成主機上的一個獨立進程。主機上的一個或者多個服務要訪問目標服務時,他們都經過同一主機上的獨立LB進程作服務發現和負載均衡。該方案也是一種分佈式方案沒有單點問題,服務調用方和LB之間是進程內調用性能好,同時該方案還簡化了服務調用方,不須要爲不一樣語言開發客戶庫。運維
本篇將介紹第二種負載均衡方法,客戶端負載均衡。
gRPC已提供了簡單的負載均衡策略(如:Round Robin),咱們只需實現它提供的Builder
和Resolver
接口,就能完成gRPC客戶端負載均衡。
type Builder interface { Build(target Target, cc ClientConn, opts BuildOption) (Resolver, error) Scheme() string }
Builder
接口:建立一個resolver
(本文稱之服務發現),用於監視名稱解析更新。
Build
方法:爲給定目標建立一個新的resolver
,當調用grpc.Dial()
時執行。
Scheme
方法:返回此resolver
支持的方案,Scheme
定義可參考:https://github.com/grpc/grpc/blob/master/doc/naming.md
type Resolver interface { ResolveNow(ResolveNowOption) Close() }
Resolver
接口:監視指定目標的更新,包括地址更新和服務配置更新。
ResolveNow
方法:被 gRPC 調用,以嘗試再次解析目標名稱。只用於提示,可忽略該方法。
Close
方法:關閉resolver
根據以上兩個接口,咱們把服務發現的功能寫在Build
方法中,把獲取到的負載均衡服務地址返回到客戶端,並監視服務更新狀況,以修改客戶端鏈接。
修改服務發現代碼,discovery.go
package etcdv3 import ( "context" "log" "sync" "time" "github.com/coreos/etcd/mvcc/mvccpb" "go.etcd.io/etcd/clientv3" "google.golang.org/grpc/resolver" ) const schema = "grpclb" //ServiceDiscovery 服務發現 type ServiceDiscovery struct { cli *clientv3.Client //etcd client cc resolver.ClientConn serverList map[string]resolver.Address //服務列表 lock sync.Mutex } //NewServiceDiscovery 新建發現服務 func NewServiceDiscovery(endpoints []string) resolver.Builder { cli, err := clientv3.New(clientv3.Config{ Endpoints: endpoints, DialTimeout: 5 * time.Second, }) if err != nil { log.Fatal(err) } return &ServiceDiscovery{ cli: cli, } } //Build 爲給定目標建立一個新的`resolver`,當調用`grpc.Dial()`時執行 func (s *ServiceDiscovery) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOption) (resolver.Resolver, error) { log.Println("Build") s.cc = cc s.serverList = make(map[string]resolver.Address) prefix := "/" + target.Scheme + "/" + target.Endpoint + "/" //根據前綴獲取現有的key resp, err := s.cli.Get(context.Background(), prefix, clientv3.WithPrefix()) if err != nil { return nil, err } for _, ev := range resp.Kvs { s.SetServiceList(string(ev.Key), string(ev.Value)) } s.cc.NewAddress(s.getServices()) //監視前綴,修改變動的server go s.watcher(prefix) return s, nil } // ResolveNow 監視目標更新 func (s *ServiceDiscovery) ResolveNow(rn resolver.ResolveNowOption) { log.Println("ResolveNow") } //Scheme return schema func (s *ServiceDiscovery) Scheme() string { return schema } //Close 關閉 func (s *ServiceDiscovery) Close() { log.Println("Close") s.cli.Close() } //watcher 監聽前綴 func (s *ServiceDiscovery) watcher(prefix string) { rch := s.cli.Watch(context.Background(), prefix, clientv3.WithPrefix()) log.Printf("watching prefix:%s now...", prefix) for wresp := range rch { for _, ev := range wresp.Events { switch ev.Type { case mvccpb.PUT: //新增或修改 s.SetServiceList(string(ev.Kv.Key), string(ev.Kv.Value)) case mvccpb.DELETE: //刪除 s.DelServiceList(string(ev.Kv.Key)) } } } } //SetServiceList 新增服務地址 func (s *ServiceDiscovery) SetServiceList(key, val string) { s.lock.Lock() defer s.lock.Unlock() s.serverList[key] = resolver.Address{Addr: val} s.cc.NewAddress(s.getServices()) log.Println("put key :", key, "val:", val) } //DelServiceList 刪除服務地址 func (s *ServiceDiscovery) DelServiceList(key string) { s.lock.Lock() defer s.lock.Unlock() delete(s.serverList, key) s.cc.NewAddress(s.getServices()) log.Println("del key:", key) } //GetServices 獲取服務地址 func (s *ServiceDiscovery) getServices() []resolver.Address { addrs := make([]resolver.Address, 0, len(s.serverList)) for _, v := range s.serverList { addrs = append(addrs, v) } return addrs }
代碼主要修改如下地方:
把獲取的服務地址轉成resolver.Address
,供gRPC客戶端鏈接。
根據schema
的定義規則,修改key
格式。
服務註冊主要修改key
存儲格式,register.go
package etcdv3 import ( "context" "log" "time" "go.etcd.io/etcd/clientv3" ) //ServiceRegister 建立租約註冊服務 type ServiceRegister struct { cli *clientv3.Client //etcd client leaseID clientv3.LeaseID //租約ID //租約keepalieve相應chan keepAliveChan <-chan *clientv3.LeaseKeepAliveResponse key string //key val string //value } //NewServiceRegister 新建註冊服務 func NewServiceRegister(endpoints []string, serName, addr string, lease int64) (*ServiceRegister, error) { cli, err := clientv3.New(clientv3.Config{ Endpoints: endpoints, DialTimeout: 5 * time.Second, }) if err != nil { log.Fatal(err) } ser := &ServiceRegister{ cli: cli, key: "/" + schema + "/" + serName + "/" + addr, val: addr, } //申請租約設置時間keepalive if err := ser.putKeyWithLease(lease); err != nil { return nil, err } return ser, nil } //設置租約 func (s *ServiceRegister) putKeyWithLease(lease int64) error { //設置租約時間 resp, err := s.cli.Grant(context.Background(), lease) if err != nil { return err } //註冊服務並綁定租約 _, err = s.cli.Put(context.Background(), s.key, s.val, clientv3.WithLease(resp.ID)) if err != nil { return err } //設置續租 按期發送需求請求 leaseRespChan, err := s.cli.KeepAlive(context.Background(), resp.ID) if err != nil { return err } s.leaseID = resp.ID s.keepAliveChan = leaseRespChan log.Printf("Put key:%s val:%s success!", s.key, s.val) return nil } //ListenLeaseRespChan 監聽 續租狀況 func (s *ServiceRegister) ListenLeaseRespChan() { for leaseKeepResp := range s.keepAliveChan { log.Println("續約成功", leaseKeepResp) } log.Println("關閉續租") } // Close 註銷服務 func (s *ServiceRegister) Close() error { //撤銷租約 if _, err := s.cli.Revoke(context.Background(), s.leaseID); err != nil { return err } log.Println("撤銷租約") return s.cli.Close() }
客戶端修改gRPC鏈接服務的部分代碼便可:
func main() { r := etcdv3.NewServiceDiscovery(EtcdEndpoints) resolver.Register(r) // 鏈接服務器 conn, err := grpc.Dial(r.Scheme()+"://8.8.8.8/simple_grpc", grpc.WithBalancerName("round_robin"), grpc.WithInsecure()) if err != nil { log.Fatalf("net.Connect err: %v", err) } defer conn.Close() // 創建gRPC鏈接 grpcClient = pb.NewSimpleClient(conn)
gRPC內置了簡單的負載均衡策略round_robin
,根據負載均衡地址,以輪詢的方式進行調用服務。
服務端啓動時,把服務地址註冊到etcd
中便可:
func main() { // 監聽本地端口 listener, err := net.Listen(Network, Address) if err != nil { log.Fatalf("net.Listen err: %v", err) } log.Println(Address + " net.Listing...") // 新建gRPC服務器實例 grpcServer := grpc.NewServer() // 在gRPC服務器註冊咱們的服務 pb.RegisterSimpleServer(grpcServer, &SimpleService{}) //把服務註冊到etcd ser, err := etcdv3.NewServiceRegister(EtcdEndpoints, SerName, Address, 5) if err != nil { log.Fatalf("register service err: %v", err) } defer ser.Close() //用服務器 Serve() 方法以及咱們的端口信息區實現阻塞等待,直到進程被殺死或者 Stop() 被調用 err = grpcServer.Serve(listener) if err != nil { log.Fatalf("grpcServer.Serve err: %v", err) } }
咱們先啓動並註冊三個服務
而後客戶端進行調用
看服務端接收到的請求
關閉localhost:8000
服務,剩餘localhost:8001
和localhost:8002
服務接收請求
從新打開localhost:8000
服務
能夠看到,gRPC客戶端負載均衡運行良好。
本文介紹了gRPC客戶端負載均衡的實現,它簡單實現了gRPC負載均衡的功能。但在對接其餘語言時候比較麻煩,須要每種語言都實現一套服務發現和負載策略,且若是要較爲複雜的負載策略,須要修改客戶端代碼才能完成。
下篇將介紹如何實現官方推薦的負載均衡策略(External Load Balancing Service
)。
源碼地址:https://github.com/Bingjian-Zhu/etcd-example
參考: