etcd分布式锁

1.实现带租约（lease）的分布式锁，如果分布式租约到期，则自动释放锁。

package main

import (
    "context"
    "fmt"
    "log"
    "os"
    "time"

    "github.com/coreos/etcd/clientv3"
    "github.com/etcd-io/etcd/clientv3/concurrency"
)

var c chan int

func init() {
    c = make(chan int)
}

func main() {
    // 创建client链接服务器，指定Config中的DialTimeout表示链接DialTimeout时间后没有链接上则返回error
    client, err := clientv3.New(clientv3.Config{
        Endpoints: []string{"http://localhost:2379"},
    })
    if err != nil {
        panic(err)
    }

    // 监听/foobar事件
    watcher := clientv3.NewWatcher(client)
    channel := watcher.Watch(context.Background(), "/foobar", clientv3.WithPrefix())
    go func() {
        for {
            select {
            case change := <-channel:
                for _, ev := range change.Events {
                    log.Printf("etcd change on key; %s, type = %v", string(ev.Kv.Key), ev.Type)
                }
            }
        }
    }()

    go lockFoobar(client, 1)
    time.Sleep(time.Second)
    go lockFoobar(client, 2)
    <-c
    <-c
}

func lockFoobar(client *clientv3.Client, id int) {
    // 创建一个10s的租约(lease)
    res, err := client.Grant(context.Background(), 10)
    if err != nil {
        panic(err)
    }
    // 利用上面创建的租约ID创建一个session
    session, err := concurrency.NewSession(client, concurrency.WithLease(res.ID))
    if err != nil {
        panic(err)
    }
    // 创建以/foobar为前缀的锁，上面监听的和这里的要相同
    mux := concurrency.NewMutex(session, "/foobar")

    log.Printf("trying to lock by #%d\n", id)
    // 获取锁使用context.Background()会一直获取锁直到获取成功
    // 如果这里使用context.WithTimeout(context.Background(), 10*time.Second)
    // 表示获取锁10s如果没有获取成功则返回error。
    if err := mux.Lock(context.Background()); err != nil {
        log.Printf("failed to lock #%d: %v\n", id, err)
        c <- id
        return
    }
    // 注意在获取锁后要调用该函数在session的租约到期后才会释放锁。
    session.Orphan()

    log.Printf("post-lock #%d (lease ID = %x) get succeed!\n", id, res.ID)
    time.Sleep(20 * time.Second)
    // 获取租约的详细信息
    //ttl, _ := client.TimeToLive(context.TODO(), res.ID)
    //log.Printf("post-post-lock-#%d-sleep. lease ttl = %v", id, ttl.TTL)
    // 这里为了测试在不释放锁的情况下，后面的的所是否能上锁成功
    // mux.Unlock(ctx)
    // log.Printf("post-unlock #%d bullshit\n", id)

    time.Sleep(200 * time.Millisecond)
    c <- id
}

session.Orphan()解释见该链接

2.实现无租约阻塞分布式锁

func main() {
    cli, err := clientv3.New(clientv3.Config{Endpoints: []string{"localhost:2379"}, DialTimeout: 5 * time.Second})
    if err != nil {
        log.Println("connect failed, err:", err)
        return
    }
    // 如果task1进程挂掉，则链接会断开那么
    //defer cli.Close()

    // ss1, err := concurrency.NewSession(cli,  concurrency.WithTTL(1))给创建的session一个1s的alive时期，即如果task1所在的进程挂掉，则ss1会被server在1s后关掉，这样task2就能立即获取成功锁
    // concurrency.WithContext(context.Background())使用该参数生成session会默认在task1所在的进程挂掉，则ss1会被server在60s后关掉，这样task2就能立即获取成功锁.
    // 所以若想让task2立即获取锁则应该使用
    // ss1, err := concurrency.NewSession(cli,  concurrency.WithTTL(1))
    ss1, err := concurrency.NewSession(cli, concurrency.WithContext(context.Background()))
    if err != nil {
        log.Fatal(err)
    }
    //defer ss1.Close()

    mu1 := concurrency.NewMutex(ss1, "/my-lock/")
    fmt.Println("try get lock1 ... ")
    // try get lock util context expire, if Lock() uses context.Background(), it will acquire lock forever
    err = mu1.Lock(context.Background())
    if err != nil {
        fmt.Println("get lock1 failed!")
        os.Exit(0)
    }

    fmt.Println("get lock1 succeed!")

    doCancel := func() {
        fmt.Println("try free lock1 ...!")
        err = mu1.Unlock(context.Background())
        if err != nil {
            //ss1.Close()
            //cli.Close()
            fmt.Println("free lock1 failed!")
            os.Exit(0)
        }
        //ss1.Close()
        //cli.Close()
        fmt.Println("free lock1 succeed!")
    }

    go func() {
        time.Sleep(15 * time.Second)
        doCancel()
    }()

    time.Sleep(2000*time.Second)
}

package main


import (
    "context"
    "fmt"
    "log"
    "os"
    "time"

    "github.com/coreos/etcd/clientv3"
    "github.com/etcd-io/etcd/clientv3/concurrency"
)

func main() {
    cli, err := clientv3.New(clientv3.Config{Endpoints: []string{"localhost:2379"}, DialTimeout: 5 * time.Second})
    if err != nil {
        log.Println("connect failed, err:", err)
        return
    }
    //defer cli.Close()

    ss2, err := concurrency.NewSession(cli,  concurrency.WithContext(context.Background()))
    if err != nil {
        log.Fatal(err)
    }
    //defer ss2.Close()

    mu2 := concurrency.NewMutex(ss2, "/my-lock/")
    fmt.Println("try get lock2 ... ")
    err = mu2.Lock(context.Background())
    if err != nil {
        fmt.Println("get lock2 failed!")
        os.Exit(0)
    }
    fmt.Println("get lock2 succeed!")

    fmt.Println("try free lock2 ...!")
    err = mu2.Unlock(context.Background())
    if err != nil {
        fmt.Println("free lock2 failed!")
        os.Exit(0)
    }
    fmt.Println("free lock2 succeed!")
    time.Sleep(2000*time.Second)
}

先运行第一个main.go，lock1获取成功。再运行第二个main.go，会一直阻塞获取lock2，这时候kill掉第一个main.go的进程，会看到第二个lock2会立即获取成功。