基於go+etcd實現分布式鎖

原文鏈接：https://www.yii-china.com/topic/detail/113

package main

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

func main() {
        var (
                config        clientv3.Config
                client        *clientv3.Client
                lease         clientv3.Lease
                leaseResp     *clientv3.LeaseGrantResponse
                leaseId       clientv3.LeaseID
                leaseRespChan <-chan *clientv3.LeaseKeepAliveResponse
                err           error
        )
        //客戶端配置
        config = clientv3.Config{
                Endpoints:   []string{"127.0.0.1:2379"},
                DialTimeout: 5 * time.Second,
        }
        //建立連接
        if client, err = clientv3.New(config); err != nil {
                fmt.Println(err)
                return
        }

        //上鎖（創建租約，自動續租）
        lease = clientv3.NewLease(client)

        //設置一個ctx取消自動續租
        ctx, cancleFunc := context.WithCancel(context.TODO())

        //設置10秒租約（過期時間）
        if leaseResp, err = lease.Grant(context.TODO(), 10); err != nil {
                fmt.Println(err)
                return
        }

        //拿到租約id
        leaseId = leaseResp.ID

        //自動續租（不停地往管道中扔租約信息）
        if leaseRespChan, err = lease.KeepAlive(ctx, leaseId); err != nil {
                fmt.Println(err)
        }

        //啟動一個協程去監聽
        go listenLeaseChan(leaseRespChan)

        //業務處理
        kv := clientv3.NewKV(client)

        //創建事務
        txn := kv.Txn(context.TODO())
        txn.If(clientv3.Compare(clientv3.CreateRevision("/cron/lock/job9"), "=", 0)).
                Then(clientv3.OpPut("/cron/lock/job9", "xxx", clientv3.WithLease(leaseId))).
                Else(clientv3.OpGet("/cron/lock/job9")) //否則搶鎖失敗

        //提交事務
        if txtResp, err := txn.Commit(); err != nil {
                fmt.Println(err)
                return
        } else {
                //判斷是否搶鎖
                if !txtResp.Succeeded {
                        fmt.Println("鎖被占用：", string(txtResp.Responses[0].GetResponseRange().Kvs[0].Value))
                        return
                }
        }

        for {
                time.Sleep(3 * time.Second)
                fmt.Println("sleep 3 s ... ...")
        }

        //釋放鎖（停止續租，終止租約）
        defer cancleFunc()                          //函數退出取消自動續租
        defer lease.Revoke(context.TODO(), leaseId) //終止租約（去掉過期時間）

        time.Sleep(10 * time.Second)
}

func listenLeaseChan(leaseRespChan <-chan *clientv3.LeaseKeepAliveResponse) {
        var (
                leaseKeepResp *clientv3.LeaseKeepAliveResponse
        )

        for {
                select {
                case leaseKeepResp = <-leaseRespChan:
                        if leaseKeepResp == nil {
                                fmt.Println("租約失效了")
                                goto END
                        } else {
                                fmt.Println("leaseKeepResp ID: ", leaseKeepResp.ID)
                        }
                }
        }
END:
}