Golang+etcd的分布式锁

新一代的后台服务中间件都用Go语言写了（上一代的才是Java），比如etcd，那么基于etcd写一个分布式，还是很实用的，废话不多说，先看代码：

package distribution_locker
 
import (
	"context"
	clientV3 "go.etcd.io/etcd/client/v3"
	"log"
	"os"
	"time"
)
 
type DistributionLocker struct {
	config         clientV3.Config
	client         *clientV3.Client
	lease          clientV3.Lease
	leaseGrantResp *clientV3.LeaseGrantResponse
	leaseId        clientV3.LeaseID
	keepRespChan   <-chan *clientV3.LeaseKeepAliveResponse
	keepResp       *clientV3.LeaseKeepAliveResponse
	txnResp        *clientV3.TxnResponse
	ctx            context.Context
	cancelFunc     context.CancelFunc
}
 
// New 创建一把锁
//  etcdEndpoints etcd连接信息，示例：[]string{"localhost:2379"}
//  connectionTimeout 连接到etcd的超时时间，示例：5*time.Second
func New(etcdEndpoints []string, connectionTimeout time.Duration) (locker *DistributionLocker, err error) {
	locker = &DistributionLocker{config: clientV3.Config{
		Endpoints:   etcdEndpoints,
		DialTimeout: connectionTimeout},
	}
	if locker.client, err = clientV3.New(locker.config); err != nil {
		return nil, err
	}
 
	//上锁并创建租约
	locker.lease = clientV3.NewLease(locker.client)
	// 第2个参数TTL，可以用于控制如果当前进程和etcd连接断开了，持有锁的上下文多长时间失效
	if locker.leaseGrantResp, err = locker.lease.Grant(context.TODO(), 3); err != nil {
		return nil, err
	}
	locker.leaseId = locker.leaseGrantResp.ID
	// 创建一个可取消的租约，主要是为了退出的时候能够释放
	locker.ctx, locker.cancelFunc = context.WithCancel(context.TODO())
 
	if locker.keepRespChan, err = locker.lease.KeepAlive(locker.ctx, locker.leaseId); err != nil {
		return nil, err
	}
	// 续约应答
	go func() {
		for {
			select {
			case locker.keepResp = <-locker.keepRespChan:
				if locker.keepRespChan == nil || locker.keepResp == nil {
					// 正式环境把这行代码注释掉，就可以避免对正常日志的干扰了
					log.Printf("进程 %+v 的锁 %+v 的租约已经失效了", os.Getpid(), locker.leaseId)
					return
				} else { // 每秒会续租一次, 所以就会收到一次应答
					// 正式环境把这行代码注释掉，就可以避免对正常日志的干扰了
					log.Printf("进程 %+v 收到自动续租应答 %+v", os.Getpid(), locker.keepResp.ID)
				}
			}
		}
	}()
	return locker, nil
}
 
func (locker *DistributionLocker) GetId() int64 {
	return int64(locker.leaseId)
}
 
// Acquire 获得锁
// lockerId 锁ID，推荐使用UUID或雪花算法，确保唯一性，防止复杂业务+大量数据的情况下发生锁冲撞
// 返回值：who 如果获得锁失败，此ID可以标示锁现在在谁手中
func (locker *DistributionLocker) Acquire(lockerId string) (who string, ok bool) {
	var err error
	// 在租约时间内去抢锁（etcd 里面的锁就是一个 key）
	kv := clientV3.NewKV(locker.client)
 
	// 创建事务
	txn := kv.Txn(context.TODO())
	// 定义锁的Key
	var lockerKey = "lock:" + lockerId
	// If 不存在 key，Then 设置它，Else 抢锁失败
	txn.If(clientV3.Compare(clientV3.CreateRevision(lockerKey), "=", 0)).
		Then(clientV3.OpPut(lockerKey, lockerId, clientV3.WithLease(locker.leaseId))).
		Else(clientV3.OpGet(lockerKey))
	if locker.txnResp, err = txn.Commit(); err != nil {
		return "", false
	}
 
	if !locker.txnResp.Succeeded {
		return string(locker.txnResp.Responses[0].GetResponseRange().Kvs[0].Value), false
	}
	return "", true
}
 
func (locker *DistributionLocker) Release() error {
	locker.cancelFunc()
	if _, err := locker.lease.Revoke(context.TODO(), locker.leaseId); err != nil {
		return err
	}
	return nil
}

核心思想是：创建一个与etcd的连接，然后在锁释放之间，不断地给锁续命，让它始终有效，和Reddsion一样的。具体地：

• 在分布式多进程环境下，锁不能设在进程中，因为两个服务是两个进程，甚至在两台计算机上
• 和集中式锁相比，分布式锁可以支持并发环境下，尤其是高并发环境下的一些数据竞争问题
• 进程1先与etcd建立连接，基于该连接设置k-v的值和版本号，形成一把锁
• 进程1对该连接不断进行续命，只要当前连接没有断开（调用release方法之前），就一直连着
• 如果进程1锁主动释放了（明确调用了release方法），连续不再续命
• 在此期间另外的进程2申请锁，与etcd建立连接
• 进程2基于该连接设置k-v的值和版本号，设置失败
• 如果进程1在锁没有释放之前意外中止了（崩溃了、被kill掉了等等），它的锁不再续命，连接被释放，其他进程此时可申请到锁

下面是测试代码：

package main
 
import (
	"distributionLockByEtcd/distribution_locker"
	"flag"
	"fmt"
	"log"
	"os"
	"time"
)
 
func main() {
	namePtr := flag.String("n", "", "锁名称")
	flag.Parse()
	if *namePtr == "" {
		fmt.Println("必须指定一个锁名称")
		return
	}
	if locker, err := distribution_locker.New([]string{"localhost:2379"}, 5*time.Second); err != nil {
		log.Fatalf("创建锁失败：%+v", err)
	} else if who, ok := locker.Acquire(*namePtr); ok {
		// 抢到锁后执行业务逻辑，没有抢到则退出
		log.Printf("进程 %+v 持有锁 %+v 正在处理任务中...", os.Getpid(), locker.GetId())
		time.Sleep(20 * time.Second) // 这是正在做的事情，假定耗时20秒
		log.Printf("进程 %+v 的任务处理完了", os.Getpid())
		// 手动释放锁，在后台应用服务中，也可以通过defer释放
		if err := locker.Release(); err != nil {
			log.Fatalf("释放锁失败：%+v", err)
		}
		time.Sleep(5 * time.Second)
	} else {
		log.Printf("获取锁失败，锁现在在 %+v 手中", who)
	}
}

相关工程已经开源，在这里 GitHub - ccpwcn/distribution_lock: 以Golang开发的基于etcd的分布式锁以Golang开发的基于etcd的分布式锁. Contribute to ccpwcn/distribution_lock development by creating an account on GitHub.https://github.com/ccpwcn/distribution_lock.git

注意：无论如何，申请锁和释放锁的操作成对出现是个好习惯，只申请不释放的做法不值得提倡。