package register
import (
"fmt"
"time"
etcd3 "github.com/coreos/etcd/clientv3"
"context"
"github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
"Test/common/service"
"log"
)
type RegisterClient struct {
etcdClient *etcd3.Client
EtcdEndpoints []string
refreshInterval time.Duration
keyTTL int
serviceName string
host string
port int
stopSignal chan bool
}
// 服务注册，命名方式"/service/{serviceName}/{serviceHost}:{servicePort}", 值为："{serviceHost}:{servicePort}"， 注册是当前服务本身而已，故可直接注册
//Tips: 如果需要，可以将值的内容转换成json格式，增减内容丰富度
func Register(serviceName string, host string, port int, EtcdEndpoints []string, refreshInterval time.Duration, keyTTL int) (*RegisterClient, error) {
client := &RegisterClient{}
client.serviceName = serviceName
client.host = host
client.port = port
client.EtcdEndpoints = EtcdEndpoints
client.refreshInterval = refreshInterval
// 保证存活时间比刷新时间长
if keyTTL < int(refreshInterval.Seconds()) {
keyTTL = 2 * int(refreshInterval.Seconds())
}
client.keyTTL = keyTTL
etcdClient, err := etcd3.New(etcd3.Config{
Endpoints: client.EtcdEndpoints,
})
if err != nil {
return nil, fmt.Errorf("Create etcd client failed: s%s", err)
}
client.stopSignal = make(chan bool, 1)
client.etcdClient = etcdClient
client.register()
return client, nil
}
func (c *RegisterClient) register() error {
go func() {
// invoke self-register with ticker
ticker := time.NewTicker(c.refreshInterval)
for {
// should get first, if not exist, set it
resp, err := c.etcdClient.Grant(context.Background(), int64(c.keyTTL))
if err != nil {
log.Printf("Register service failed. serviceName: %s, err: %s", c.serviceName, err.Error())
}
fmt.Println(c.getKey())
_, err = c.etcdClient.Get(context.Background(), c.getKey())
value := fmt.Sprintf("%s:%d", c.host, c.port)
if err != nil {
if err == rpctypes.ErrKeyNotFound {
if _, err := c.etcdClient.Put(context.Background(), c.getKey(), value, etcd3.WithLease(resp.ID)); err != nil {
//if _, err := c.etcdClient.Put(context.Background(), c.getKey(), value); err != nil {
log.Printf("Register service failed. serviceName: %s, value: %s, err: %s", c.serviceName, value, err.Error())
}
} else {
log.Printf("Get Value from etcd failed: %s", err)
}
} else {
if _, err := c.etcdClient.Put(context.Background(), c.getKey(), value, etcd3.WithLease(resp.ID)); err != nil {
//if _, err := c.etcdClient.Put(context.Background(), c.getKey(), value); err != nil {
log.Printf("Register service failed. serviceName: %s, value: %s, err: %s", c.serviceName, value, err.Error())
} else {
log.Printf("Register current service: %s, value: %s", c.serviceName, value)
}
}
select {
case <-c.stopSignal: // 手动停止注册
return
case <-ticker.C: // 等待一定时间间隔
}
}
}()
return nil
}
// 只有服务在注册之后，才能够取消注册
func (c *RegisterClient) UnRegister() error {
c.stopSignal <- true
_, err := c.etcdClient.Get(context.Background(), c.getKey())
if err != nil {
log.Printf("Get Value from etcd failed: %s", err.Error())
} else if _, err := c.etcdClient.Delete(context.Background(), c.getKey()); err != nil {
log.Printf("UnRegister service failed. serviceName: %s, err: %s", c.serviceName, err.Error())
}
return err
}
func (c *RegisterClient) getKey() string {
return fmt.Sprintf("/%s/%s/%s:%d", service.Prefix, c.serviceName, c.host, c.port)
}
/* Copyright 2018 Bruce Liu. All rights reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
*
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package discovery
import (
etcd3 "github.com/coreos/etcd/clientv3"
"fmt"
"errors"
"math/rand"
"context"
"Test/common/service"
"github.com/coreos/etcd/mvcc/mvccpb"
"log"
)
// 基本方式：本地维护一个服务发现的结果，优先从本地服务读取值. 第一次获取一个服务的地址时，因本地没有，故从etcd服务获取，然后换到到本地，并同时启动该服务的watcher，有更新时，及时更新本地缓存。
// Tips 一个服务需要发现多个其他服务，故需要有client，且后期通过client来获取服务的值
type DiscoveryClient struct {
etcdClient *etcd3.Client
EtcdEndpoints []string
// 服务列表: key 为service 名字 todo 本地缓存也需要有一个时效，以及一个反馈服务失效的地方，这个可以后期做服务降级之类的东西的时候再做
serviceMap map[string][]string
watcher map[string]bool
}
func NewClient(etcdEndpoints []string) (*DiscoveryClient, error) {
client := &DiscoveryClient{}
client.EtcdEndpoints = etcdEndpoints
etcdClient, err := etcd3.New(etcd3.Config{
Endpoints: client.EtcdEndpoints,
})
if err != nil {
return nil, fmt.Errorf("Create register etcdClient failed: %s", err.Error())
}
client.etcdClient = etcdClient
client.serviceMap = make(map[string][]string)
client.watcher = make(map[string]bool)
return client, nil
}
func (c *DiscoveryClient) GetService(serviceName string) (string, error) {
if serviceName == "" {
return "", errors.New("Service name should not be null")
}
// 从本地缓存的获取
if len(c.serviceMap[serviceName]) > 0 {
i := rand.Intn(len(c.serviceMap[serviceName]))
return c.serviceMap[serviceName][i], nil
}
// 从 etcd服务获取
resp, err := c.etcdClient.Get(context.Background(), c.getKey(serviceName), etcd3.WithPrefix())
if err != nil {
log.Printf("Get service failed: %s, error: %s", serviceName, err.Error())
return "", err
}
// 建立watcher，有变化，及时更新 只启动一个watch?
if !c.watcher[serviceName] {
go c.watch(serviceName)
}
serviceList := service.ExtractAddress(resp)
if len(serviceList) == 0 {
return "", errors.New("Service not found.")
}
c.serviceMap[serviceName] = serviceList
i := rand.Intn(len(c.serviceMap[serviceName]))
return c.serviceMap[serviceName][i], nil
}
func (c *DiscoveryClient) getKey(serviceName string) string {
return fmt.Sprintf("/%s/%s", service.Prefix, serviceName)
}
// 开始watch这个service的注册信息
func (c *DiscoveryClient) watch(serviceName string) {
c.watcher[serviceName] = true
defer func() {
// 如果此函数退出，那么则可以重新建立watcher
c.watcher[serviceName] = false
}()
// prefix is the etcd prefix/value to watch
prefix := c.getKey(serviceName)
log.Printf("Watch service : %s", serviceName)
// 创建watch
rch := c.etcdClient.Watch(context.Background(), prefix, etcd3.WithPrefix())
for wresp := range rch {
for _, ev := range wresp.Events {
log.Printf("Catch changes: %s, event: %d", serviceName, ev.Type)
switch ev.Type {
case mvccpb.PUT:
find := false
for _, v := range c.serviceMap[serviceName] {
if v == string(ev.Kv.Value) {
find = true
break
}
}
if !find {
log.Printf("Add new service: %s, value: %s", serviceName, string(ev.Kv.Value))
c.serviceMap[serviceName] = append(c.serviceMap[serviceName], string(ev.Kv.Value))
}
break
case mvccpb.DELETE:
index := -1
key := string(ev.Kv.Key)
value := key[len(c.getKey(serviceName))+1:len(key)]
for i, v := range c.serviceMap[serviceName] {
if v == value {
index = i
break
}
}
if index >= 0 && index < len(c.serviceMap[serviceName]) {
log.Printf("Delete service: %s, value: %s", serviceName, string(ev.Kv.Value))
c.serviceMap[serviceName] = append(c.serviceMap[serviceName][0:index], c.serviceMap[serviceName][index+1: len(c.serviceMap[serviceName])]...)
}
break
}
}
}
}
