美文网首页Go知识库微服务
go语言基于etcd的服务注册与发现

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

作者: xyt001 | 来源:发表于2018-12-15 13:01 被阅读177次

    前言:

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

    服务注册与发现的必要:

    讲一个很简单的场景,一般服务端架构最前面是 一台网关 ,网关后面是 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

    相关文章

      网友评论

        本文标题:go语言基于etcd的服务注册与发现

        本文链接:https://www.haomeiwen.com/subject/mwynhqtx.html