go任務調度7（etcd的watch的用法）

監聽etcd中的kv變化，常用來做集群中的配置下發、狀態同步，非常有價值。

package main

import (
    "go.etcd.io/etcd/clientv3"
    "time"
    "fmt"
    "context"
    "go.etcd.io/etcd/mvcc/mvccpb"
)

func main() {
    var (
        config clientv3.Config
        client *clientv3.Client
        err error
        kv clientv3.KV
        watcher clientv3.Watcher
        getResp *clientv3.GetResponse
        watchStartRevision int64
        watchRespChan <-chan clientv3.WatchResponse
        watchResp clientv3.WatchResponse
        event *clientv3.Event
    )

    // 客戶端配置
    config = clientv3.Config{
        Endpoints: []string{"0.0.0.0:2379"},
        DialTimeout: 5 * time.Second,
    }

    // 建立連接
    if client, err = clientv3.New(config); err != nil {
        fmt.Println(err)
        return
    }

    // KV
    kv = clientv3.NewKV(client)

    // 模擬etcd中KV的變化
    go func() {
        for {
            kv.Put(context.TODO(), "/cron/jobs/job7", "i am job7")

            kv.Delete(context.TODO(), "/cron/jobs/job7")

            time.Sleep(1 * time.Second)
        }
    }()

    // 先GET到當前的值，并監聽后續變化
    if getResp, err = kv.Get(context.TODO(), "/cron/jobs/job7"); err != nil {
        fmt.Println(err)
        return
    }

    // 現在key是存在的
    if len(getResp.Kvs) != 0 {
        fmt.Println("當前值:", string(getResp.Kvs[0].Value))
    }

    // 當前etcd集群事務ID, 單調遞增的（監聽/cron/jobs/job7后續的變化,也就是通過監聽版本變化）
    watchStartRevision = getResp.Header.Revision + 1

    // 創建一個watcher(監聽器)
    watcher = clientv3.NewWatcher(client)

    // 啟動監聽
    fmt.Println("從該版本向后監聽:", watchStartRevision)

    ctx, cancelFunc := context.WithCancel(context.TODO())
    //5秒鐘后取消
    time.AfterFunc(5 * time.Second, func() {
        cancelFunc()
    })
    //這里ctx感知到cancel則會關閉watcher
    watchRespChan = watcher.Watch(ctx, "/cron/jobs/job7", clientv3.WithRev(watchStartRevision))

    // 處理kv變化事件
    for watchResp = range watchRespChan {
        for _, event = range watchResp.Events {
            switch event.Type {
            case mvccpb.PUT:
                fmt.Println("修改為:", string(event.Kv.Value), "Revision:", event.Kv.CreateRevision, event.Kv.ModRevision)
            case mvccpb.DELETE:
                fmt.Println("刪除了", "Revision:", event.Kv.ModRevision)
            }
        }
    }
}

輸出：

[root@bogon etcd]# go run demo8.go
當前值: i am job7
從該版本向后監聽: 33
刪除了 Revision: 33
修改為: i am job7 Revision: 34 34
刪除了 Revision: 35
修改為: i am job7 Revision: 36 36
刪除了 Revision: 37
修改為: i am job7 Revision: 38 38
刪除了 Revision: 39
修改為: i am job7 Revision: 40 40
刪除了 Revision: 41
[root@bogon etcd]#