go基於grpc構建微服務框架-服務註冊與發現

概述

grpc 是谷歌開源的rpc框架,基於http2實現,並支持跨語言,目前基本涵蓋了主流語言.跨語言的實現主要得益於protobuf,經過編寫proto文件,經過protobuf工具生成對應語言的類庫進行使用.

對於go這樣一門新生語言來講,生態鏈還處於發展階段,微服務框架也是如此,下面將基於grpc-go版本搭建一個微服務通信框架.

1.服務註冊與發佈的機制

1.1 解決的問題

服務註冊與發佈主要解決的服務依賴問題,一般意義上,若是A服務調用B服務時,最直接的作法是配置IP地址和端口.但隨着服務依賴變多時,配置將會十分龐雜,且當服務發生遷移時,那麼全部相關服務的配置均須要修改,這將十分難以維護以及容易出現問題. 所以爲了解決這種服務依賴關係,服務註冊與發佈應運而生.

1.2 機制

服務註冊與發現主要分爲如下幾點.

• 服務信息發佈
  這裏主要是服務的服務名,IP信息,以及一些附件元數據.經過註冊接口註冊到服務註冊發佈中心.
• 存活檢測
  當服務意外中止時,客戶端須要感知到服務中止,並將服務的IP地址踢出可用的IP地址列表,這裏可使用定時心跳去實現.
• 客戶端負載均衡
  經過服務註冊與發佈,能夠實現一個服務部署多臺實例,客戶端實如今實例直接的負載均衡,從而實現服務的橫向擴展.

所以,服務註冊與發佈能夠概況爲,服務將信息上報,客戶端拉取服務信息,經過服務名進行調用,當服務宕機時客戶端踢掉故障服務,服務新上線時客戶端自動添加到調用列表.

2.實現

grpc-go的整個實現大量使用go的接口特性,所以經過擴展接口,能夠很容易的實現服務的註冊與發現,這裏服務註冊中心考慮到可用性以及一致性,通常採用etcd或zookeeper來實現,這裏實現etcd的版本.
完整代碼以及使用示例見:grpc-wrapper

2.1 客戶端

具體須要實現幾個接口,針對客戶端,最簡單的實現方式只須要實現兩個接口方法Resolve(),以及Next(),而後使用輪詢的負載均衡方式.
主要經過etcd的Get接口以及Watch接口實現.

• Resolve()接口

//用於生成Watcher,監聽註冊中心中的服務信息變化
func (er *etcdRegistry) Resolve(target string) (naming.Watcher, error) {
	ctx, cancel := context.WithTimeout(context.TODO(), resolverTimeOut)
	w := &etcdWatcher{
		cli:    er.cli,
		target: target + "/",
		ctx:    ctx,
		cancel: cancel,
	}
	return w, nil
}

複製代碼

• Next() 接口

//Next接口主要用於獲取註冊的服務信息,經過channel以及watch,當服務信息發生
//變化時,Next接口會將變化返回給grpc框架從而實現服務信息變動.
func (ew *etcdWatcher) Next() ([]*naming.Update, error) {
	var updates []*naming.Update
    //初次獲取時,建立監聽channel,並返回獲取到的服務信息
	if ew.watchChan == nil {
		//create new chan
		resp, err := ew.cli.Get(ew.ctx, ew.target, etcd.WithPrefix(), etcd.WithSerializable())
		if err != nil {
			return nil, err
		}
		for _, kv := range resp.Kvs {
			var upt naming.Update
			if err := json.Unmarshal(kv.Value, &upt); err != nil {
				continue
			}
			updates = append(updates, &upt)
		}
        //建立etcd的watcher監聽target(服務名)的信息.
		opts := []etcd.OpOption{etcd.WithRev(resp.Header.Revision + 1), etcd.WithPrefix(), etcd.WithPrevKV()}
		ew.watchChan = ew.cli.Watch(context.TODO(), ew.target, opts...)
		return updates, nil
	}

    //阻塞監聽,服務發生變化時才返回給上層
	wrsp, ok := <-ew.watchChan
	if !ok {
		err := status.Error(codes.Unavailable, "etcd watch closed")
		return nil, err
	}
	if wrsp.Err() != nil {
		return nil, wrsp.Err()
	}
	for _, e := range wrsp.Events {
		var upt naming.Update
		var err error
		switch e.Type {
		case etcd.EventTypePut:
			err = json.Unmarshal(e.Kv.Value, &upt)
			upt.Op = naming.Add
		case etcd.EventTypeDelete:
			err = json.Unmarshal(e.PrevKv.Value, &upt)
			upt.Op = naming.Delete
		}

		if err != nil {
			continue
		}
		updates = append(updates, &upt)
	}
	return updates, nil
}

複製代碼

2.2 服務端

服務端只須要上報服務信息,並定時保持心跳,這裏經過etcd的Put接口以及KeepAlive接口實現. 具體以下:

func (er *etcdRegistry) Register(ctx context.Context, target string, update naming.Update, opts ...wrapper.RegistryOptions) (err error) {
	//將服務信息序列化成json格式
    var upBytes []byte
	if upBytes, err = json.Marshal(update); err != nil {
		return status.Error(codes.InvalidArgument, err.Error())
	}

	ctx, cancel := context.WithTimeout(context.TODO(), resolverTimeOut)
	er.cancal = cancel
	rgOpt := wrapper.RegistryOption{TTL: wrapper.DefaultRegInfTTL}
	for _, opt := range opts {
		opt(&rgOpt)
	}

	switch update.Op {
	case naming.Add:
		lsRsp, err := er.lsCli.Grant(ctx, int64(rgOpt.TTL/time.Second))
		if err != nil {
			return err
		}

        //Put服務信息到etcd,並設置key的值TTL,經過後面的KeepAlive接口
        //對TTL進行續期,超過TTL的時間未收到續期請求,則說明服務可能掛了,從而清除服務信息
		etcdOpts := []etcd.OpOption{etcd.WithLease(lsRsp.ID)}
		key := target + "/" + update.Addr
		_, err = er.cli.KV.Put(ctx, key, string(upBytes), etcdOpts...)
		if err != nil {
			return err
		}

        //保持心跳
		lsRspChan, err := er.lsCli.KeepAlive(context.TODO(), lsRsp.ID)
		if err != nil {
			return err
		}
		go func() {
			for {
				_, ok := <-lsRspChan
				if !ok {
					grpclog.Fatalf("%v keepalive channel is closing", key)
					break
				}
			}
		}()
	case naming.Delete:
		_, err = er.cli.Delete(ctx, target+"/"+update.Addr)
	default:
		return status.Error(codes.InvalidArgument, "unsupported op")
	}
	return nil
}
複製代碼

3. 參考

grpc
etcd
grpc-wrapper