go语言基于etcd的服务注册与发现

前言：

一般随着服务流量的增多，单台服务器是不可能满足我们的需求了 ，这个时候我们 就需要 部署很多台 运行着 一样服务的 机器 ，这个就是 简单的分布式了。

服务注册与发现的必要：

讲一个很简单的场景，一般服务端架构最前面是 一台网关 ，网关后面是 n 台运行着一样的 服务 的机器。 客户端一般就是访问网关 ，然后 网关 就把流量 转发到 后面的 服务器上。那么我们来考虑这么一个问题，后面 服务器 信息处理不过来的时候，我们需要加机器。最low的方法就是 加上一台服务器，然后 修改网关服务器的配置表 加上 新加的 服务器的IP 和端口，然后重启网关。还有就是 当 后面的 服务器万一 down 了，网关是不知道的，还会把 流量转发到 down的 服务器上，造成 服务的不可用。

etcd的简单介绍：

因为前面说的问题我们就需要etcd 这类产品了，etcd 是一个 分布式 的高一致性的 键值存储系统。我们每次网关 后面加一个服务，只需要向etcd 注册 该服务（其实就是 存一个值）然后向etcd 发送心跳，当etcd 没有检测到心跳就会 把这个键值对 删了(这整个动作是etcd里的租约模式)，网关那边 就只需要 watch 这个 key ，就能够知道 所有服务的所有动态了。

服务注册的简单实现：

网上找了蛮多资料后，加上自己封装后的产物。开箱即用。

package main

import (
    "context"
    "fmt"
    "go.etcd.io/etcd/clientv3"
    "time"
)

//创建租约注册服务
type ServiceReg struct {
    client        *clientv3.Client
    lease         clientv3.Lease
    leaseResp     *clientv3.LeaseGrantResponse
    canclefunc    func()
    keepAliveChan <-chan *clientv3.LeaseKeepAliveResponse
    key           string
}

func NewServiceReg(addr []string, timeNum int64) (*ServiceReg, error) {
    conf := clientv3.Config{
        Endpoints:   addr,
        DialTimeout: 5 * time.Second,
    }

    var (
        client *clientv3.Client
    )

    if clientTem, err := clientv3.New(conf); err == nil {
        client = clientTem
    } else {
        return nil, err
    }

    ser := &ServiceReg{
        client: client,
    }

    if err := ser.setLease(timeNum); err != nil {
        return nil, err
    }
    go ser.ListenLeaseRespChan()
    return ser, nil
}

//设置租约
func (this *ServiceReg) setLease(timeNum int64) error {
    lease := clientv3.NewLease(this.client)

    //设置租约时间
    leaseResp, err := lease.Grant(context.TODO(), timeNum)
    if err != nil {
        return err
    }

    //设置续租
    ctx, cancelFunc := context.WithCancel(context.TODO())
    leaseRespChan, err := lease.KeepAlive(ctx, leaseResp.ID)

    if err != nil {
        return err
    }

    this.lease = lease
    this.leaseResp = leaseResp
    this.canclefunc = cancelFunc
    this.keepAliveChan = leaseRespChan
    return nil
}

//监听 续租情况
func (this *ServiceReg) ListenLeaseRespChan() {
    for {
        select {
        case leaseKeepResp := <-this.keepAliveChan:
            if leaseKeepResp == nil {
                fmt.Printf("已经关闭续租功能\n")
                return
            } else {
                fmt.Printf("续租成功\n")
            }
        }
    }
}

//通过租约 注册服务
func (this *ServiceReg) PutService(key, val string) error {
    kv := clientv3.NewKV(this.client)
    _, err := kv.Put(context.TODO(), key, val, clientv3.WithLease(this.leaseResp.ID))
    return err
}


//撤销租约
func (this *ServiceReg) RevokeLease() error {
    this.canclefunc()
    time.Sleep(2 * time.Second)
    _, err := this.lease.Revoke(context.TODO(), this.leaseResp.ID)
    return err
}

func main() {
    ser,_ := NewServiceReg([]string{"127.0.0.1:2379"},5)
    ser.PutService("/node/111","heiheihei")
    select{}
}

那就让我来简单的解释一下吧。

1. etcd的 租约模式:客户端申请 一个租约 并设置 过期时间，每隔一段时间 就要 请求 etcd 申请续租。客户端可以通过租约存key。如果不续租 ，过期了，etcd 会删除这个租约上的 所有key-value。类似于心跳模式。
2. 一般相同的服务存的 key 的前缀是一样的 比如 “server/001"=> "127.0.0.1:1212" 和 ”server/002"=>"127.0.0.1:1313" 这种模式，然后 客户端 就直接 匹配 “server/” 这个key。
3. 具体代码不复杂。要是看不懂就留言问我吧。

服务发现的简单实现：

import (
    "go.etcd.io/etcd/clientv3"
    "time"
    "context"
    "go.etcd.io/etcd/mvcc/mvccpb"
    "sync"
    "log"
)

type ClientDis struct {
    client        *clientv3.Client
    serverList    map[string]string
    lock          sync.Mutex
}

func NewClientDis (addr []string)( *ClientDis, error){
    conf := clientv3.Config{
        Endpoints:   addr,
        DialTimeout: 5 * time.Second,
    }
    if client, err := clientv3.New(conf); err == nil {
        return &ClientDis{
            client:client,
            serverList:make(map[string]string),
        }, nil
    } else {
        return nil ,err
    }
}


func (this * ClientDis) GetService(prefix string) ([]string ,error){
    resp, err := this.client.Get(context.Background(), prefix, clientv3.WithPrefix())
    if err != nil {
        return nil, err
    }
    addrs := this.extractAddrs(resp)

    go this.watcher(prefix)
    return addrs ,nil
}


func (this *ClientDis) watcher(prefix string) {
    rch := this.client.Watch(context.Background(), prefix, clientv3.WithPrefix())
    for wresp := range rch {
        for _, ev := range wresp.Events {
            switch ev.Type {
            case mvccpb.PUT:
                this.SetServiceList(string(ev.Kv.Key),string(ev.Kv.Value))
            case mvccpb.DELETE:
                this.DelServiceList(string(ev.Kv.Key))
            }
        }
    }
}

func (this *ClientDis) extractAddrs(resp *clientv3.GetResponse) []string {
    addrs := make([]string,0)
    if resp == nil || resp.Kvs == nil {
        return addrs
    }
    for i := range resp.Kvs {
        if v := resp.Kvs[i].Value; v != nil {
            this.SetServiceList(string(resp.Kvs[i].Key),string(resp.Kvs[i].Value))
            addrs = append(addrs, string(v))
        }
    }
    return addrs
}

func (this *ClientDis) SetServiceList(key,val string) {
    this.lock.Lock()
    defer this.lock.Unlock()
    this.serverList[key] = string(val)
    log.Println("set data key :",key,"val:",val)
}

func (this *ClientDis) DelServiceList(key string) {
    this.lock.Lock()
    defer this.lock.Unlock()
    delete(this.serverList,key)
    log.Println("del data key:", key)
}


func (this *ClientDis) SerList2Array()[]string {
    this.lock.Lock()
    defer this.lock.Unlock()
    addrs := make([]string,0)

    for _, v := range this.serverList {
        addrs = append(addrs,v)
    }
    return addrs
}

func main () {
    cli,_ := NewClientDis([]string{"127.0.0.1:2379"})
    cli.GetService("/node")
    select {}
}

1.创建一个client 连到etcd。
2.匹配到所有相同前缀的 key。把值存到 serverList 这个map里面。
3 watch这个 key前缀，当有增加或者删除的时候 就 修改这个map。
4所以这个map就是 实时的 服务列表

总结

以上就是基于 etcd的服务注册与发现了，大家可以 自己动手试试，载一个 etcd ，然后分别运行这两个文件。就能看到效果了。 代码我也传到github上面了 https://github.com/mistaker/etcdTool 。