parent
633489b480
commit
54fe4e84dd
@ -0,0 +1,31 @@
|
||||
package gojobs
|
||||
|
||||
import (
|
||||
"context"
|
||||
"go.etcd.io/etcd/client/v3"
|
||||
)
|
||||
|
||||
// Watch 监听
|
||||
func (e Etcd) Watch(ctx context.Context, key string, opts ...clientv3.OpOption) clientv3.WatchChan {
|
||||
return e.client.Watch(ctx, key, opts...)
|
||||
}
|
||||
|
||||
// Create 创建
|
||||
func (e Etcd) Create(ctx context.Context, key, val string, opts ...clientv3.OpOption) (*clientv3.PutResponse, error) {
|
||||
return e.client.Put(ctx, key, val, opts...)
|
||||
}
|
||||
|
||||
// Get 获取
|
||||
func (e Etcd) Get(ctx context.Context, key string, opts ...clientv3.OpOption) (*clientv3.GetResponse, error) {
|
||||
return e.client.Get(ctx, key, opts...)
|
||||
}
|
||||
|
||||
// Update 更新
|
||||
func (e Etcd) Update(ctx context.Context, key, val string, opts ...clientv3.OpOption) (*clientv3.PutResponse, error) {
|
||||
return e.client.Put(ctx, key, val, opts...)
|
||||
}
|
||||
|
||||
// Delete 删除
|
||||
func (e Etcd) Delete(ctx context.Context, key string, opts ...clientv3.OpOption) (*clientv3.DeleteResponse, error) {
|
||||
return e.client.Delete(ctx, key, opts...)
|
||||
}
|
||||
@ -0,0 +1,60 @@
|
||||
package gojobs
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"go.etcd.io/etcd/api/v3/mvccpb"
|
||||
"go.etcd.io/etcd/client/v3"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// NewEtcdServer 创建 etcd server
|
||||
func NewEtcdServer(config *EtcdConfig) (e *Etcd, err error) {
|
||||
|
||||
e.Endpoints = config.Endpoints
|
||||
e.DialTimeout = config.DialTimeout
|
||||
|
||||
e.client, err = clientv3.New(clientv3.Config{
|
||||
Endpoints: e.Endpoints,
|
||||
DialTimeout: e.DialTimeout,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, errors.New("连接失败：" + err.Error())
|
||||
}
|
||||
|
||||
// 得到KV和Lease的API子集
|
||||
e.kv = clientv3.NewKV(e.client)
|
||||
e.lease = clientv3.NewLease(e.client)
|
||||
|
||||
return e, nil
|
||||
}
|
||||
|
||||
// ListWorkers 获取在线worker列表
|
||||
func (e Etcd) ListWorkers() (workerArr []string, err error) {
|
||||
var (
|
||||
getResp *clientv3.GetResponse
|
||||
kv *mvccpb.KeyValue
|
||||
workerIP string
|
||||
)
|
||||
|
||||
// 初始化数组
|
||||
workerArr = make([]string, 0)
|
||||
|
||||
// 获取目录下所有Kv
|
||||
if getResp, err = e.kv.Get(context.TODO(), JobWorkerDir, clientv3.WithPrefix()); err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// 解析每个节点的IP
|
||||
for _, kv = range getResp.Kvs {
|
||||
// kv.Key : /cron/workers/192.168.2.1
|
||||
workerIP = ExtractWorkerIP(string(kv.Key))
|
||||
workerArr = append(workerArr, workerIP)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// ExtractWorkerIP 提取worker的IP
|
||||
func ExtractWorkerIP(regKey string) string {
|
||||
return strings.TrimPrefix(regKey, JobWorkerDir)
|
||||
}
|
||||
@ -0,0 +1,101 @@
|
||||
package gojobs
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"go.dtapp.net/goip"
|
||||
"go.etcd.io/etcd/client/v3"
|
||||
"log"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
// JobWorkerDir 服务注册目录
|
||||
JobWorkerDir = "/cron/workers/"
|
||||
)
|
||||
|
||||
// NewEtcdWorker 创建 etcd Worker
|
||||
func NewEtcdWorker(config *EtcdConfig) (e *Etcd, err error) {
|
||||
|
||||
e.Endpoints = config.Endpoints
|
||||
e.DialTimeout = config.DialTimeout
|
||||
|
||||
e.client, err = clientv3.New(clientv3.Config{
|
||||
Endpoints: e.Endpoints,
|
||||
DialTimeout: e.DialTimeout,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, errors.New("连接失败：" + err.Error())
|
||||
}
|
||||
|
||||
// 得到KV和Lease的API子集
|
||||
e.kv = clientv3.NewKV(e.client)
|
||||
e.lease = clientv3.NewLease(e.client)
|
||||
|
||||
go e.RegisterWorker()
|
||||
|
||||
return e, nil
|
||||
}
|
||||
|
||||
// RegisterWorker 注册worker
|
||||
func (e Etcd) RegisterWorker() {
|
||||
var (
|
||||
regKey string
|
||||
leaseGrantResp *clientv3.LeaseGrantResponse
|
||||
err error
|
||||
keepAliveChan <-chan *clientv3.LeaseKeepAliveResponse
|
||||
keepAliveResp *clientv3.LeaseKeepAliveResponse
|
||||
cancelCtx context.Context
|
||||
cancelFunc context.CancelFunc
|
||||
)
|
||||
|
||||
localIP := goip.GetOutsideIp()
|
||||
|
||||
for {
|
||||
// 注册路径
|
||||
regKey = JobWorkerDir + localIP
|
||||
|
||||
cancelFunc = nil
|
||||
|
||||
// 创建租约
|
||||
leaseGrantResp, err = e.lease.Grant(context.TODO(), 10)
|
||||
if err != nil {
|
||||
log.Println("创建租约")
|
||||
goto RETRY
|
||||
}
|
||||
|
||||
// 自动续租
|
||||
keepAliveChan, err = e.lease.KeepAlive(context.TODO(), leaseGrantResp.ID)
|
||||
if err != nil {
|
||||
log.Println("自动续租")
|
||||
goto RETRY
|
||||
}
|
||||
|
||||
cancelCtx, cancelFunc = context.WithCancel(context.TODO())
|
||||
|
||||
// 注册到etcd
|
||||
_, err = e.kv.Put(cancelCtx, regKey, "", clientv3.WithLease(leaseGrantResp.ID))
|
||||
if err != nil {
|
||||
log.Println(fmt.Sprintf(" %s 服务注册失败:%s", regKey, err))
|
||||
goto RETRY
|
||||
}
|
||||
|
||||
// 处理续租应答
|
||||
for {
|
||||
select {
|
||||
case keepAliveResp = <-keepAliveChan:
|
||||
if keepAliveResp == nil { // 续租失败
|
||||
log.Println("续租失败")
|
||||
goto RETRY
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
RETRY:
|
||||
time.Sleep(1 * time.Second)
|
||||
if cancelFunc != nil {
|
||||
cancelFunc()
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in new issue