- 获取在线worker列表

etcd.go

@@ -2,7 +2,10 @@ package gojobs
 
 import (
 	"context"
+	"errors"
+	"go.etcd.io/etcd/api/v3/mvccpb"
 	"go.etcd.io/etcd/client/v3"
+	"strings"
 	"time"
 )
 
@@ -15,54 +18,95 @@ type EtcdConfig struct {
 // Etcd etcd
 type Etcd struct {
 	EtcdConfig                  // 配置
-	c          *clientv3.Client // 驱动
+	client     *clientv3.Client // 驱动
+	kv         clientv3.KV
+	lease      clientv3.Lease
 }
 
 // Close 关闭
 func (e Etcd) Close() {
-	e.c.Close()
+	e.client.Close()
 }
 
 // NewEtcd 创建etcd
-func NewEtcd(config *EtcdConfig) *Etcd {
+func NewEtcd(config *EtcdConfig) (*Etcd, error) {
 
 	e := &Etcd{}
 	e.Endpoints = config.Endpoints
 	e.DialTimeout = config.DialTimeout
 
 	var err error
-	e.c, err = clientv3.New(clientv3.Config{
+	e.client, err = clientv3.New(clientv3.Config{
 		Endpoints:   e.Endpoints,
 		DialTimeout: e.DialTimeout,
 	})
 	if err != nil {
-		panic("连接失败：" + err.Error())
+		return nil, errors.New("连接失败：" + err.Error())
 	}
 
-	return e
+	// 得到KV和Lease的API子集
+	e.kv = clientv3.NewKV(e.client)
+	e.lease = clientv3.NewLease(e.client)
+
+	return e, nil
 }
 
 // Watch 监听
 func (e Etcd) Watch(ctx context.Context, key string, opts ...clientv3.OpOption) clientv3.WatchChan {
-	return e.c.Watch(ctx, key, opts...)
+	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.c.Put(ctx, key, val, opts...)
+	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.c.Get(ctx, key, opts...)
+	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.c.Put(ctx, key, val, opts...)
+	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.c.Delete(ctx, key, opts...)
+	return e.client.Delete(ctx, key, opts...)
+}
+
+// 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
+}
+
+const (
+	// JobWorkerDir 服务注册目录
+	JobWorkerDir = "/cron/workers/"
+)
+
+// ExtractWorkerIP 提取worker的IP
+func ExtractWorkerIP(regKey string) string {
+	return strings.TrimPrefix(regKey, JobWorkerDir)
 }

version.go

@@ -1,3 +1,3 @@
 package gojobs
 
-const Version = "1.0.6"
+const Version = "1.0.7"