etcd实现分布式锁

转载自：etcd实现分布式锁

三台网站制作公司哪家好，找创新互联公司！从网页设计、网站建设、微信开发、APP开发、响应式网站设计等网站项目制作，到程序开发，运营维护。创新互联公司于2013年成立到现在10年的时间，我们拥有了丰富的建站经验和运维经验，来保证我们的工作的顺利进行。专注于网站建设就选创新互联公司。

当并发的访问共享资源的时候，如果没有加锁的话，无法保证共享资源安全性和正确性。这个时候就需要用到锁

1、需要具备的特性

需要保证互斥访问（分布式环境需要保证不同节点、不同线程的互斥访问）
需要有超时机制，防止锁意外未释放，其他节点无法获取到锁；也要保证任务能够正常执行完成，不能超时了任务还没结束，导致任务执行一般被释放锁
需要有阻塞和非阻塞两种请求锁的接口

2、本地锁

当业务执行在同一个线程内，也就是我初始化一个本地锁，其他请求也认这把锁。一般是服务部署在单机环境下。

我们可以看下下面的例子，开1000个goroutine并发的给Counter做自增操作，结果会是什么样的呢？

package main

import (
	"fmt"
	"sync"
)

var sg sync.WaitGroup

type Counter struct {
	count int
}

// 自增操作
func (m *Counter) Incr() {
	m.count++
}

// 获取总数
func (m *Counter) Count() int {
	return m.count
}

func main() {
	c := &Counter{}
	for i := 0; i < 1000; i++ {
		sg.Add(1)
    // 模拟并发请求
		go func() {
			c.Incr()
			sg.Done()
		}()
	}
	sg.Wait()

	fmt.Println(c.Count())
}

结果是count的数量并不是预想中的1000，而是下面这样，每次打印出的结果都不一样，但是接近1000

user@userdeMacBook-Pro  ~/go/src/go-demo/mutex  go run main.go
953
 user@userdeMacBook-Pro  ~/go/src/go-demo/mutex  go run main.go
982
 user@userdeMacBook-Pro  ~/go/src/go-demo/mutex  go run main.go
984

出现这个问题的原因就是没有给自增操作加锁

下面我们修改代码如下，在Incr中加上go的mutex互斥锁

package main

import (
	"fmt"
	"sync"
)

var sg sync.WaitGroup

type Counter struct {
	count int
	mu    sync.Mutex
}

func (m *Counter) Incr() {
  // 每次写之前先加锁，写完之后释放锁
	m.mu.Lock()
	defer m.mu.Unlock()
	m.count++
}

func (m *Counter) Count() int {
	return m.count
}

func main() {
	c := &Counter{}
	for i := 0; i < 1000; i++ {
		sg.Add(1)
		go func() {
			c.Incr()
			sg.Done()
		}()
	}
	sg.Wait()

	fmt.Println(c.Count())
}

可以看到现在count正常输出1000了

user@userdeMacBook-Pro  ~/go/src/go-demo/mutex  go run main.go
1000
 user@userdeMacBook-Pro  ~/go/src/go-demo/mutex  go run main.go
1000
 user@userdeMacBook-Pro  ~/go/src/go-demo/mutex  go run main.go
1000

3、etcd分布式锁

简单部署一个etcd集群

├── docker-compose.yml
├── etcd
│   └── Dockerfile

Dockerfile文件内容

FROM bitnami/etcd:latest

LABEL maintainer="liuyuede123 "

Docker-compose.yml内容

version: '3.5'
# 网络配置
networks:
  backend:
    driver: bridge

# 服务容器配置
services:
  etcd1:                                  # 自定义容器名称
    build:
      context: etcd                    # 指定构建使用的 Dockerfile 文件
    environment:
      - TZ=Asia/Shanghai
      - ALLOW_NONE_AUTHENTICATION=yes
      - ETCD_NAME=etcd1
      - ETCD_INITIAL_ADVERTISE_PEER_URLS=http://etcd1:2380
      - ETCD_LISTEN_PEER_URLS=http://0.0.0.0:2380
      - ETCD_LISTEN_CLIENT_URLS=http://0.0.0.0:2379
      - ETCD_ADVERTISE_CLIENT_URLS=http://etcd1:2379
      - ETCD_INITIAL_CLUSTER_TOKEN=etcd-cluster
      - ETCD_INITIAL_CLUSTER=etcd1=http://etcd1:2380,etcd2=http://etcd2:2380,etcd3=http://etcd3:2380
      - ETCD_INITIAL_CLUSTER_STATE=new
    ports:                               # 设置端口映射
      - ":2379"
      - ":2380"
    networks:
      - backend
    restart: always

  etcd2: # 自定义容器名称
    build:
      context: etcd                    # 指定构建使用的 Dockerfile 文件
    environment:
      - TZ=Asia/Shanghai
      - ALLOW_NONE_AUTHENTICATION=yes
      - ETCD_NAME=etcd2
      - ETCD_INITIAL_ADVERTISE_PEER_URLS=http://etcd2:2380
      - ETCD_LISTEN_PEER_URLS=http://0.0.0.0:2380
      - ETCD_LISTEN_CLIENT_URLS=http://0.0.0.0:2379
      - ETCD_ADVERTISE_CLIENT_URLS=http://etcd2:2379
      - ETCD_INITIAL_CLUSTER_TOKEN=etcd-cluster
      - ETCD_INITIAL_CLUSTER=etcd1=http://etcd1:2380,etcd2=http://etcd2:2380,etcd3=http://etcd3:2380
      - ETCD_INITIAL_CLUSTER_STATE=new
    ports: # 设置端口映射
      - ":2379"
      - ":2380"
    networks:
      - backend
    restart: always

  etcd3: # 自定义容器名称
    build:
      context: etcd                    # 指定构建使用的 Dockerfile 文件
    environment:
      - TZ=Asia/Shanghai
      - ALLOW_NONE_AUTHENTICATION=yes
      - ETCD_NAME=etcd3
      - ETCD_INITIAL_ADVERTISE_PEER_URLS=http://etcd3:2380
      - ETCD_LISTEN_PEER_URLS=http://0.0.0.0:2380
      - ETCD_LISTEN_CLIENT_URLS=http://0.0.0.0:2379
      - ETCD_ADVERTISE_CLIENT_URLS=http://etcd3:2379
      - ETCD_INITIAL_CLUSTER_TOKEN=etcd-cluster
      - ETCD_INITIAL_CLUSTER=etcd1=http://etcd1:2380,etcd2=http://etcd2:2380,etcd3=http://etcd3:2380
      - ETCD_INITIAL_CLUSTER_STATE=new
    ports: # 设置端口映射
      - ":2379"
      - ":2380"
    networks:
      - backend
    restart: always

执行docker-compose up -d启动etcd服务，可以看到docker中已经启动了3个服务

实现互斥访问

package main

import (
	"fmt"
	clientv3 "go.etcd.io/etcd/client/v3"
	"go.etcd.io/etcd/client/v3/concurrency"
	"sync"
)

var sg sync.WaitGroup

type Counter struct {
	count int
}

func (m *Counter) Incr() {
	m.count++
}

func (m *Counter) Count() int {
	return m.count
}

func main() {
	endpoints := []string{"http://127.0.0.1:", "http://127.0.0.1:", "http://127.0.0.1:"}
  // 初始化etcd客户端
	client, err := clientv3.New(clientv3.Config{Endpoints: endpoints})
	if err != nil {
		fmt.Println(err)
		return
	}
	defer client.Close()

	counter := &Counter{}

	sg.Add(100)
	for i := 0; i < 100; i++ {
		go func() {
      // 这里会生成租约，默认是60秒
			session, err := concurrency.NewSession(client)
			if err != nil {
				panic(err)
			}
			defer session.Close()

			locker := concurrency.NewLocker(session, "/my-test-lock")
			locker.Lock()
			counter.Incr()
			locker.Unlock()
			sg.Done()
		}()
	}
	sg.Wait()

	fmt.Println("count:", counter.Count())
}

执行结果：

user@userdeMacBook-Pro  ~/go/src/go-demo/mutex  go run main.go
count: 100
 user@userdeMacBook-Pro  ~/go/src/go-demo/mutex  go run main.go
count: 100
 user@userdeMacBook-Pro  ~/go/src/go-demo/mutex  go run main.go
count: 100

实现超时机制

当某个客户端持有锁时，由于某些原因导致锁未释放，就会导致这个客户端一直持有这把锁，其他客户端一直获取不到锁。所以需要分布式锁实现超时机制，当锁未释放时，会因为etcd的租约会到期而释放锁。当业务正常处理时，租约到期之前会继续续约，知道业务处理完毕释放锁。

package main

import (
	"fmt"
	clientv3 "go.etcd.io/etcd/client/v3"
	"go.etcd.io/etcd/client/v3/concurrency"
	"sync"
	"time"
)

var sg sync.WaitGroup

type Counter struct {
	count int
}

func (m *Counter) Incr() {
	m.count++
}

func (m *Counter) Count() int {
	return m.count
}

func main() {
	endpoints := []string{"http://127.0.0.1:", "http://127.0.0.1:", "http://127.0.0.1:"}
	client, err := clientv3.New(clientv3.Config{Endpoints: endpoints})
	if err != nil {
		fmt.Println(err)
		return
	}
	defer client.Close()

	counter := &Counter{}

	session, err := concurrency.NewSession(client)
	if err != nil {
		panic(err)
	}
	defer session.Close()

	locker := concurrency.NewLocker(session, "/my-test-lock")
	fmt.Println("locking...", time.Now().Format("2006-01-02 15:04:05"))
	locker.Lock()
	fmt.Println("locked...", time.Now().Format("2006-01-02 15:04:05"))
  // 模拟业务
	time.Sleep(100 * time.Second)
	counter.Incr()
	locker.Unlock()
	fmt.Println("released...", time.Now().Format("2006-01-02 15:04:05"))

	fmt.Println("count:", counter.Count())
}

命令行开2个窗口，第一个窗口执行程序并获取锁，之后模拟意外退出并没有调用unlock方法

go run main.go
locking... 2022-09-03 23:41:48 # 租约生成时间
locked... 2022-09-03 23:41:48
^Csignal: interrupt

第二个窗口，在第一个窗口退出之前尝试获取锁，此时是阻塞状态。第一个窗口退出之后由于租约还没到期，第二个窗口还是获取锁的状态。等到第一个窗口租约到期（默认60秒），第二个获取锁成功

locking... 2022-09-03 23:41:52
locked... 2022-09-03 23:42:48 # 第一个租约60秒到期，获取锁成功
released... 2022-09-03 23:44:28
count: 1

实现阻塞和非阻塞接口

上面的例子中已经实现了阻塞接口，即当前有获取到锁的请求，则其他请求阻塞等待锁释放

非阻塞的方式就是尝试获取锁，如果失败立即返回。etcd中是实现了tryLock方法

// TryLock locks the mutex if not already locked by another session.
// If lock is held by another session, return immediately after attempting necessary cleanup
// The ctx argument is used for the sending/receiving Txn RPC.
func (m *Mutex) TryLock(ctx context.Context) error {

具体看下面的例子

package main

import (
	"context"
	"fmt"
	clientv3 "go.etcd.io/etcd/client/v3"
	"go.etcd.io/etcd/client/v3/concurrency"
	"sync"
	"time"
)

var sg sync.WaitGroup

type Counter struct {
	count int
}

func (m *Counter) Incr() {
	m.count++
}

func (m *Counter) Count() int {
	return m.count
}

func main() {
	endpoints := []string{"http://127.0.0.1:", "http://127.0.0.1:", "http://127.0.0.1:"}
	client, err := clientv3.New(clientv3.Config{Endpoints: endpoints})
	if err != nil {
		fmt.Println(err)
		return
	}
	defer client.Close()

	counter := &Counter{}

	session, err := concurrency.NewSession(client)
	if err != nil {
		panic(err)
	}
	defer session.Close()

  // 此处使用newMutex初始化
	locker := concurrency.NewMutex(session, "/my-test-lock")
	fmt.Println("locking...", time.Now().Format("2006-01-02 15:04:05"))
	err = locker.TryLock(context.Background())
  // 获取锁失败就抛错
	if err != nil {
		fmt.Println("lock failed", err)
		return
	}
	fmt.Println("locked...", time.Now().Format("2006-01-02 15:04:05"))
	time.Sleep(100 * time.Second)
	counter.Incr()
	err = locker.Unlock(context.Background())
	if err != nil {
		fmt.Println("unlock failed", err)
		return
	}
	fmt.Println("released...", time.Now().Format("2006-01-02 15:04:05"))

	fmt.Println("count:", counter.Count())
}

窗口1、窗口2执行结果

go run main.go
locking... 2022-09-04 00:00:21
locked... 2022-09-04 00:00:21
released... 2022-09-04 00:02:01
count: 1

go run main.go
locking... 2022-09-04 00:00:27
lock failed mutex: Locked by another session

新闻标题：etcd实现分布式锁
地址分享：http://chengdu.cdxwcx.cn/article/dsoicdo.html

甜橘子，专注成都网站制作网站设计与营销型网站建设与优化

首页

网站建设

网站制作案例

解决方案

网站设计报价

网站制作动态

关于我们

联系我们

成都网站建设设计将想法与焦点和您一起共享

etcd实现分布式锁

1、需要具备的特性

2、本地锁

3、etcd分布式锁

简单部署一个etcd集群

实现互斥访问

实现超时机制

实现阻塞和非阻塞接口

其他资讯

mysql怎么免配置,mysql高可用配置

怎么在电脑搭建oracle环境,linux搭建oracle

CSS样式说法不正确的,css样式中的选择符说法正确的是

移动端jquery,移动端什么意思

html5企业模板,html电商模板

甜橘子，专注成都网站制作网站设计与营销型网站建设与优化

成都网站建设设计 将想法与焦点和您一起共享

etcd实现分布式锁

1、需要具备的特性

2、本地锁

3、etcd分布式锁

简单部署一个etcd集群

实现互斥访问

实现超时机制

实现阻塞和非阻塞接口

其他资讯

mysql怎么免配置,mysql高可用配置

怎么在电脑搭建oracle环境,linux搭建oracle

CSS样式说法不正确的,css样式中的选择符说法正确的是

移动端jquery,移动端什么意思

html5企业模板,html电商模板

成都网站建设设计将想法与焦点和您一起共享