系統中實現服務註冊與發現所需的基本功能有node
在分佈式系統中,如何管理節點間的狀態一直是一個難題,etcd 是由開發並維護的,它使用 Go 語言編寫,並經過Raft 一致性算法處理日誌複製以保證強一致性。etcd像是專門爲集羣環境的服務發現和註冊而設計,它提供了數據 TTL 失效、數據改變監視、多值、目錄監聽、分佈式鎖原子操做等功能,能夠方便的跟蹤並管理集羣節點的狀態。算法
咱們寫兩個 Demo 程序,一個服務充當service,一個客戶端程序充當網關代理。服務運行後會去etcd 以本身服務名命名的目錄中註冊服務節點,並定時續租(更新 TTL)。客戶端從 etcd查詢服務目錄中的節點信息代理服務的請求,而且會在協程中實時監控服務目錄中的變化,維護到本身的服務節點信息列表中。mvc
// 將服務註冊到etcd上 func RegisterServiceToETCD(ServiceTarget string, value string) { dir = strings.TrimRight(ServiceTarget, "/") + "/" client, err := clientv3.New(clientv3.Config{ Endpoints: []string{"localhost:2379"}, DialTimeout: 5 * time.Second, }) if err != nil { panic(err) } kv := clientv3.NewKV(client) lease := clientv3.NewLease(client) var curLeaseId clientv3.LeaseID = 0 for { if curLeaseId == 0 { leaseResp, err := lease.Grant(context.TODO(), 10) if err != nil { panic(err) } key := ServiceTarget + fmt.Sprintf("%d", leaseResp.ID) if _, err := kv.Put(context.TODO(), key, value, clientv3.WithLease(leaseResp.ID)); err != nil { panic(err) } curLeaseId = leaseResp.ID } else { // 續約租約,若是租約已通過期將curLeaseId復位到0從新走建立租約的邏輯 if _, err := lease.KeepAliveOnce(context.TODO(), curLeaseId); err == rpctypes.ErrLeaseNotFound { curLeaseId = 0 continue } } time.Sleep(time.Duration(1) * time.Second) } }
type HelloService struct {} func (p *HelloService) Hello(request string, reply *string) error { *reply = "hello:" + request return nil } var serviceTarget = "Hello" var port = ":1234" var host = "remote_host"// 僞代碼 func main() { rpc.RegisterName("HelloService", new(HelloService)) listener, err := net.Listen("tcp", port) if err != nil { log.Fatal("ListenTCP error:", err) } conn, err := listener.Accept() if err != nil { log.Fatal("Accept error:", err) } go RegisterServiceToETCD(serviceTarget, host + port) rpc.ServeConn(conn) }
網關經過 etcd獲取到服務目錄下的全部節點的信息,將他們初始化到自身維護的可訪問服務節點列表中。而後使用Watch機制監聽etcd上服務對應的目錄的更新,根據通道發送過來的PUT和DELETE事件來增長和刪除服務的可用節點列表。tcp
var serviceTarget = "Hello" type remoteService struct { name string nodes map[string]string mutex sync.Mutex } // 獲取服務目錄下全部key初始化到服務的可用節點列表中 func getService(etcdClient clientv3.Client) *remoteService { service = &remoteService { name: serviceTarget } kv := clientv3.NewKV(etcdClient) rangeResp, err := kv.Get(context.TODO(), service.name, clientv3.WithPrefix()) if err != nil { panic(err) } service.mutex.Lock() for _, kv := range rangeResp.Kvs { service.nodes[string(kv.Key)] = string(kv.Value) } service.mutex.Unlock() go watchServiceUpdate(etcdClient, service) } // 監控服務目錄下的事件 func watchServiceUpdate(etcdClient clientv3.Client, service *remoteService) { watcher := clientv3.NewWatcher(client) // Watch 服務目錄下的更新 watchChan := watcher.Watch(context.TODO(), service.name, clientv3.WithPrefix()) for watchResp := range watchChan { for _, event := range watchResp.Events { service.mutex.Lock() switch (event.Type) { case mvccpb.PUT://PUT事件,目錄下有了新key service.nodes[string(event.Kv.Key)] = string(event.Kv.Value) case mvccpb.DELETE://DELETE事件,目錄中有key被刪掉(Lease過時,key 也會被刪掉) delete(service.nodes, string(event.Kv.Key)) } service.mutex.Unlock() } } } func main () { client, err := clientv3.New(clientv3.Config{ Endpoints: []string{"remote_host:2379"}, DialTimeout: 5 * time.Second, }) service := getService(client)// 獲取服務的可用節點 ...... // 每次有請求過來從服務節點中選取一個鏈接,而後給節點發送請求 rpcClient, _ = rpc.Dial("tcp", service.nodes[i]) var reply string rpcClient.Call("HelloService.hello", &reply) ...... }
除了上面說的客戶端或者網關發現系統中的已存服務外,系統中的各個服務之間也須要感知到其餘角色的存在,服務間的發現方法與上面的例子相似,每一個服務都能做爲客戶端在 etcd 中發現其餘服務的存在。分佈式
說明:程序爲便於理解有不少僞代碼,主要是說明思路,想要實際運行起來還須要不少編碼工做,歡迎有這方面經驗的朋友交流想法。編碼