[etcd lock by golang]

etcd_lock.go

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
    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, cancleFunc := context.WithCancel(context.TODO())

  if leaseResp, err = lease.Grant(context.TODO(), 10); err !=nil {
    fmt.Println(err)
    return
  }
  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(cotext.TODO())
  //创建事务
  txn.If(clientv3.Compare(clientv3.CreateRevision("/cron/lock/job"), "=", 0)).
      Then(clientv3.OpPut("/cron/lock/job", "locked", clientv3.WithLease(leaseId))).Else(clientv3.OpGet("/cron/lock/job")) //否则抢锁失败
  
  //提交事务
  if txnResp, err := txn.Commit(); err != nil {
    fmt.Println(err)
    return
  } else {
    if !txnResp.Succeeded {
      fmt.Println("锁被占用: ", string(txnResp.Responses[0].GetResponseRange().Kvs[0].Value))
      return
    }
  }
  fmt.Println("处理任务")

  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)
        }
    }
  }
  END:
}