golang etcd服务注册与发现

清疚 2023-02-20 02:10 20阅读 0赞

#### 产生场景 ####

服务端架构最前面是 一台网关节点 ，网关后面是多台运行着一样的 服务的计算节点，客户端一般就是访问网关 ，然后 网关 就把流量 转发到 后面的计算节点上。如果后面的计算节点信息处理不过来的时候，我们需要加机器，最low的方法就是 加上一台服务器，然后 修改网关服务器的配置表 加上 新加的 服务器的IP 和端口，然后重启网关。但如果后面的机器一旦down 了，网关是不知道的，还会把流量转发到 down的机器上，造成服务的不可用，etcd可以避免硬件的单点故障或网络问题

#### etcd注册 ####

etcd 是一个 分布式键值存储系统，在一个集群中，如果一个节点配置了某些属性，集群中的每个节点都可以使用完整的存档，我们每次在网关节点后的计算节点集群加一个服务，只需要向etcd 注册 该服务（其实就是 存一个值）然后向etcd 发送心跳，当etcd 没有检测到心跳就会 把这个键值对 删了(这整个动作是etcd里的租约模式)，网关那边 就只需要 watch 这个 key ，就能够知道 所有服务的所有动态了

##### 启动服务端： #####

./etcd --listen-client-urls ‘http://0.0.0.0:2379’ --advertise-client-urls ‘http://0.0.0.0:2379’  
–listen-client-urls：监听的用于客户端通信的url,可以监听多个  
–advertise-client-urls：建议使用的客户端通信url

##### etcdctl #####

etcdctl是官方提供的客户端程序。  
etcd有两个版本的接口，v2和v3，且两个版本不兼容，v2已经停止了支持，v3性能更好。etcdctl默认使用v2版本，如果想使用v3版本，可通过环境变量ETCDCTL\_API=3进行设置

# 设置环境变量
    > export ETCDCTL_API=3
    
    # 放入一个 键值对
    > ./etcdctl put "name" "amber"
    OK
    
    # 取出 key
    > ./etcdctl get name
    name
    amber
    
    # 删除 key
    > ./etcdctl del name
    1
    
    # 再获取name
    > ./etcdctl get name
    # 什么都没拿到
    
    > ./etcdctl put "/project/jobs/job1" "job1"
    OK
    
    > ./etcdctl put "/project/jobs/job2" "job2"
    OK
    
    # 获取某个目录下的所有key,但是直接获取某个目录 etcd是不认识的 因为它没有目录的概念
    > ./etcdctl get "/project/jobs/"
    # 需要加上 --prefix命令
    > ./etcdctl get "/project/jobs/" --prefix
    /project/jobs/job1
    job1
    /project/jobs/job2
    job2
    
    # 删除所有
    > ./etcdctl del "/project/jobs/" --prefix
    2
    
    # 开始监听某个key
    > ./etcdctl watch "name"
    # 。。。什么都不打印，在等待结果
    
    # 当另一个窗口执行了 ./etcdctl put "name" "amber" 之后
    $ ./etcdctl watch "name"
    PUT
    name
    amber
    
    # 当另一个窗口执行了 ./etcdctl del "name" 之后
    $ ./etcdctl watch "name"
    PUT
    name
    amber
    DELETE
    name

###### 续租 ######

# 创建一个20s的租约
    > ./etcdctl lease grant 20
    lease 694d673115905e37 granted with TTL(20s)
    
    # 使用租约的 id 进行 put 操作
    > ./etcdctl put --lease=694d673115905e37 "name" "amber"
    
    # 20s后get发现 key被删除了
    > ./etcdctl get "name"
    # 空应答
    
    # 删除租约
    > ./etcdctl lease revoke 694d673115905e49
    lease 694d673115905e49 revoked
    
    # 自动续租
    > ./etcdctl lease keep-alive 694d673115905e4f
    lease 694d673115905e4f keepalived with TTL(20)
    lease 694d673115905e4f keepalived with TTL(20)

###### golang实现etcd续租 ######

import (
        "context"
        "fmt"
        "go.etcd.io/etcd/clientv3"
        "time"
    )
    
    type Service struct { 
        client *clientv3.Client
    }
    
    // 指定client端，Endpoints是etcd server的机器列表，DialTimeout是计算节点链接服务的超时时间
    func NewService() *Service { 
        config := clientv3.Config{ 
            Endpoints:   []string{ "106.12.72.181:23791", "106.12.72.181:23792"},
            DialTimeout: 10 * time.Second,
        }
        client, _ := clientv3.New(config)
        return &Service{ client: client}
    }
    
    //注册服务
    func (this *Service) RegService(id string, name string, address string) error { 
        kv := clientv3.NewKV(this.client)
        key_prefix := "/etcd_services/"
        ctx := context.Background()
        lease := clientv3.NewLease(this.client)
        
        //设置租约过期时间为20秒
        leaseRes, err := clientv3.NewLease(this.client).Grant(ctx, 20) 
        if err != nil { 
            return err
        }
        _, err = kv.Put(context.Background(), key_prefix+id+"/"+name, address, clientv3.WithLease(leaseRes.ID)) //把服务的key绑定到租约下面
        if err != nil { 
            return err
        }
        //续租时间大概自动为租约的三分之一时间，context.TODO官方定义为是你不知道要传什么
        keepaliveRes, err := lease.KeepAlive(context.TODO(), leaseRes.ID) context的时候就用这个
        if err != nil { 
            return err
        }
        go lisKeepAlive(keepaliveRes)
        return err
    }
    
    func lisKeepAlive(keepaliveRes <-chan *clientv3.LeaseKeepAliveResponse) { 
        for { 
            select { 
            case ret := <-keepaliveRes:
                if ret != nil { 
                    fmt.Println("续租成功", time.Now())
                }
            }
        }
    }

###### golang server实现etcd监听 ######

package main
    
    import (
    	"go.etcd.io/etcd/clientv3"
    	"time"
    	"context"
    	"go.etcd.io/etcd/mvcc/mvccpb"
    	"sync"
    	"log"
    )
    
    type ClientMon struct { 
    	client        *clientv3.Client
    	serverList    map[string]string
    	lock          sync.Mutex
    }
    
    // 初始化server端
    func NewClientMon (addr []string)( *ClientMon, error){ 
    	conf := clientv3.Config{ 
    		Endpoints:   addr,
    		DialTimeout: 5 * time.Second,
    	}
    	if client, err := clientv3.New(conf); err == nil { 
    		return &ClientMon{ 
    			client:client,
    			serverList:make(map[string]string),
    		}, nil
    	} else { 
    		return nil ,err
    	}
    }
    
    // 获取prefix目录下所有内容，并返回
    func (this * ClientMon) GetService(prefix string) ([]string ,error){ 
    	resp, err := this.client.Get(context.Background(), prefix, clientv3.WithPrefix())
    	if err != nil { 
    		return nil, err
    	}
    	
    	// resp是从指定prefix服务下get回的value，extractAddrs将value内容存到list
    	addrs := this.extractAddrs(resp)
    
        go this.watcher(prefix)
    	return addrs ,nil
    }
    
    // 将获取到的prefix目录下所有内容存入list并返回
    func (this *ClientMon) 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
    }
    
    // watch负责将监听到的put、delete请求存放到指定list
    func (this *ClientMon) 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 *ClientMon) 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 *ClientMon) DelServiceList(key string) { 
    	this.lock.Lock()
    	defer this.lock.Unlock()
    	delete(this.serverList,key)
    	log.Println("del data key:", key)
    }
    
    func main () { 
    	client,_ := NewClientMon([]string{ "127.0.0.1:2379"})
    	client.GetService("/node")
    }