1.實(shí)現(xiàn)帶租約（lease）的分布式鎖什黑，如果分布式租約到期削锰，則自動(dòng)釋放鎖妹卿。
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() {
// 創(chuàng)建client鏈接服務(wù)器渣玲，指定Config中的DialTimeout表示鏈接DialTimeout時(shí)間后沒(méi)有鏈接上則返回error
client, err := clientv3.New(clientv3.Config{
Endpoints: []string{"http://localhost:2379"},
})
if err != nil {
panic(err)
}
// 監(jiān)聽(tīng)/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) {
// 創(chuàng)建一個(gè)10s的租約(lease)
res, err := client.Grant(context.Background(), 10)
if err != nil {
panic(err)
}
// 利用上面創(chuàng)建的租約ID創(chuàng)建一個(gè)session
session, err := concurrency.NewSession(client, concurrency.WithLease(res.ID))
if err != nil {
panic(err)
}
// 創(chuàng)建以/foobar為前綴的鎖逗概，上面監(jiān)聽(tīng)的和這里的要相同
mux := concurrency.NewMutex(session, "/foobar")
log.Printf("trying to lock by #%d\n", id)
// 獲取鎖使用context.Background()會(huì)一直獲取鎖直到獲取成功
// 如果這里使用context.WithTimeout(context.Background(), 10*time.Second)
// 表示獲取鎖10s如果沒(méi)有獲取成功則返回error。
if err := mux.Lock(context.Background()); err != nil {
log.Printf("failed to lock #%d: %v\n", id, err)
c <- id
return
}
// 注意在獲取鎖后要調(diào)用該函數(shù)在session的租約到期后才會(huì)釋放鎖逾苫。
session.Orphan()
log.Printf("post-lock #%d (lease ID = %x) get succeed!\n", id, res.ID)
time.Sleep(20 * time.Second)
// 獲取租約的詳細(xì)信息
//ttl, _ := client.TimeToLive(context.TODO(), res.ID)
//log.Printf("post-post-lock-#%d-sleep. lease ttl = %v", id, ttl.TTL)
// 這里為了測(cè)試在不釋放鎖的情況下枚钓，后面的的所是否能上鎖成功
// mux.Unlock(ctx)
// log.Printf("post-unlock #%d bullshit\n", id)
time.Sleep(200 * time.Millisecond)
c <- id
}
session.Orphan()解釋見(jiàn)該鏈接
2.實(shí)現(xiàn)無(wú)租約阻塞分布式鎖
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進(jìn)程掛掉铅搓，則鏈接會(huì)斷開(kāi)那么
//defer cli.Close()
// ss1, err := concurrency.NewSession(cli, concurrency.WithTTL(1))給創(chuàng)建的session一個(gè)1s的alive時(shí)期，即如果task1所在的進(jìn)程掛掉秘噪，則ss1會(huì)被server在1s后關(guān)掉狸吞，這樣task2就能立即獲取成功鎖
// concurrency.WithContext(context.Background())使用該參數(shù)生成session會(huì)默認(rèn)在task1所在的進(jìn)程掛掉指煎，則ss1會(huì)被server在60s后關(guān)掉便斥，這樣task2就能立即獲取成功鎖.
// 所以若想讓task2立即獲取鎖則應(yīng)該使用
// 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)
}
先運(yùn)行第一個(gè)main.go枢纠，lock1獲取成功像街。再運(yùn)行第二個(gè)main.go晋渺，會(huì)一直阻塞獲取lock2，這時(shí)候kill掉第一個(gè)main.go的進(jìn)程木西，會(huì)看到第二個(gè)lock2會(huì)立即獲取成功。
