• 基于Redis Cluster的分布式锁实现以互斥方式操作共享资源

    今天要说的技术方案也是有一定项目背景的。在上一个项目中，我们需要对一个redis集群中过期的key进行处理，这是一个分布式

    系统，考虑到高可用性，需要具备过期处理功能的服务有多个副本，这样我们就要求在同一时间内仅有一个副本可以对过期的key>进行处理，如果该副本挂掉，系统会在其他副本中再挑选出一个来处理过期的key。

    很显然，这里涉及到一个选主(leader election)的过程。每当涉及选主，很多人就会想到一些高大上的分布式一致性/共识算法，

    比如： raft 、 paxos 等。当然使用这些算法自然没有问题，但是也给系统徒增了很多复杂性。能否有一些更简单直接的方案呢？我们已经有了一个redis集群，是否可>以利用redis集群的能力来完成这一点呢？

    Redis原生并没有提供leader election算法，但Redis作者提供了 分布式锁的算法 ，也就是说我们可以用分布式锁来实现一个简单的选主功能，见下图：

     

    在上图中我们看到，只有持有锁的服务才具备操作数据的资格，也就是说持有锁的服务的角色是leader，而其他服务则继续尝试去持有锁，它们是follower的角色。

    1. 基于单节点redis的分布式锁

    在redis官方 有关分布式锁算法的介绍页面 中，作者给出了各种编程语言的推荐实现，而Go语言的推荐实现仅 redsync 这一种。在这篇短文中，我们就来使用redsync实现基于Redis分布式锁的选主方案。

    在Go生态中，连接和操作redis的主流go客户端库有 go-redis 和 redigo 。最新的redsync版本底层redis driver既支持go-redis，也支持redigo，我个人日常使用最多的是go-redis这个客户端，这里我们就用go-redis。

    redsync github主页中给出的例子是基于单redis node的分布式锁示例。下面我们也先以单redis节点来看看如何通过Redis的分布式锁实现我们的业务逻辑：

    1. // github.com/bigwhite/experiments/blob/master/redis-cluster-distributed-lock/standalone/main.go
    2.  
    3.      1  package main
    4.      2
    5.      3  import (
    6.      4      "context"
    7.      5      "log"
    8.      6      "os"
    9.      7      "os/signal"
    10.      8      "sync"
    11.      9      "sync/atomic"
    12.     10      "syscall"
    13.     11      "time"
    14.     12
    15.     13      goredislib "github.com/go-redis/redis/v8"
    16.     14      "github.com/go-redsync/redsync/v4"
    17.     15      "github.com/go-redsync/redsync/v4/redis/goredis/v8"
    18.     16  )
    19.     17
    20.     18  const (
    21.     19      redisKeyExpiredEventSubj = `__keyevent@0__:expired`
    22.     20  )
    23.     21
    24.     22  var (
    25.     23      isLeader  int64
    26.     24      m         atomic.Value
    27.     25      id        string
    28.     26      mutexName = "the-year-of-the-ox-2021"
    29.     27  )
    30.     28
    31.     29  func init() {
    32.     30      if len(os.Args) < 2 {
    33.     31          panic("args number is not correct")
    34.     32      }
    35.     33      id = os.Args[1]
    36.     34  }
    37.     35
    38.     36  func tryToBecomeLeader() (bool, func() (bool, error), error) {
    39.     37      client := goredislib.NewClient(&goredislib.Options{
    40.     38          Addr: "localhost:6379",
    41.     39      })
    42.     40      pool := goredis.NewPool(client)
    43.     41      rs := redsync.New(pool)
    44.     42
    45.     43      mutex := rs.NewMutex(mutexName)
    46.     44
    47.     45      if err := mutex.Lock(); err != nil {
    48.     46          client.Close()
    49.     47          return false, nil, err
    50.     48      }
    51.     49
    52.     50      return true, func() (bool, error) {
    53.     51          return mutex.Unlock()
    54.     52      }, nil
    55.     53  }
    56.     54
    57.     55  func doElectionAndMaintainTheStatus(quit <-chan struct{}) {
    58.     56      ticker := time.NewTicker(time.Second * 5)
    59.     57      var err error
    60.     58      var ok bool
    61.     59      var cf func() (bool, error)
    62.     60
    63.     61      c := goredislib.NewClient(&goredislib.Options{
    64.     62          Addr: "localhost:6379",
    65.     63      })
    66.     64      defer c.Close()
    67.     65      for {
    68.     66          select {
    69.     67          case <-ticker.C:
    70.     68              if atomic.LoadInt64(&isLeader) == 0 {
    71.     69                  ok, cf, err = tryToBecomeLeader()
    72.     70                  if ok {
    73.     71                      log.Printf("prog-%s become leader successfully\n", id)
    74.     72                      atomic.StoreInt64(&isLeader, 1)
    75.     73                      defer cf()
    76.     74                  }
    77.     75                  if !ok || err != nil {
    78.     76                      log.Printf("prog-%s try to become leader failed: %s\n", id, err)
    79.     77                  }
    80.     78              } else {
    81.     79                  log.Printf("prog-%s is the leader\n", id)
    82.     80                  // update the lock live time and maintain the leader status
    83.     81                  c.Expire(context.Background(), mutexName, 8*time.Second)
    84.     82              }
    85.     83          case <-quit:
    86.     84              return
    87.     85          }
    88.     86      }
    89.     87  }
    90.     88
    91.     89  func doExpire(quit <-chan struct{}) {
    92.     90      // subscribe the expire event of redis
    93.     91      c := goredislib.NewClient(&goredislib.Options{
    94.     92          Addr: "localhost:6379"})
    95.     93      defer c.Close()
    96.     94
    97.     95      ctx := context.Background()
    98.     96      pubsub := c.Subscribe(ctx, redisKeyExpiredEventSubj)
    99.     97      _, err := pubsub.Receive(ctx)
    100.     98      if err != nil {
    101.     99          log.Printf("prog-%s subscribe expire event failed: %s\n", id, err)
    102.    100          return
    103.    101      }
    104.    102      log.Printf("prog-%s subscribe expire event ok\n", id)
    105.    103
    106.    104      // Go channel which receives messages from redis db
    107.    105      ch := pubsub.Channel()
    108.    106      for {
    109.    107          select {
    110.    108          case event := <-ch:
    111.    109              key := event.Payload
    112.    110              if atomic.LoadInt64(&isLeader) == 0 {
    113.    111                  break
    114.    112              }
    115.    113              log.Printf("prog-%s 收到并处理一条过期消息[key:%s]", id, key)
    116.    114          case <-quit:
    117.    115              return
    118.    116          }
    119.    117      }
    120.    118  }
    121.    119
    122.    120  func main() {
    123.    121      var wg sync.WaitGroup
    124.    122      wg.Add(2)
    125.    123      var quit = make(chan struct{})
    126.    124
    127.    125      go func() {
    128.    126          doElectionAndMaintainTheStatus(quit)
    129.    127          wg.Done()
    130.    128      }()
    131.    129      go func() {
    132.    130          doExpire(quit)
    133.    131          wg.Done()
    134.    132      }()
    135.    133
    136.    134      c := make(chan os.Signal, 1)
    137.    135      signal.Notify(c, syscall.SIGINT, syscall.SIGTERM)
    138.    136      _ = <-c
    139.    137      close(quit)
    140.    138      log.Printf("recv exit signal...")
    141.    139      wg.Wait()
    142.    140      log.Printf("program exit ok")
    143.    141  }

    上面示例代码比较长，但它很完整。我们一点点来看。

    首先，我们看 120~141行 的main函数结构。在这个函数中，我们创建了两个新goroutine，main goroutine通过sync.WaitGroup等待这两个子goroutine的退出并使用quit channel模式(关于goroutine的并发模式的详解，可以参考我的专栏文章 《Go并发模型和常见并发模式》 )在收到系统信号(关于signal包的使用，请参见我的专栏文章 《小心被kill！不要忽略对系统信号的处理》 )后通知两个子goroutine退出。

    接下来，我们逐个看两个子goroutine的执行逻辑。第一个goroutine执行的是doElectionAndMaintainTheStatus函数。该函数会持续尝试去持有分布式锁(tryToBecomeLeader)，一旦持有，它就变成了分布式系统中的leader角色；成为leader角色的副本会保持其角色状态(见81行)。

    尝试持有分布式锁并成为leader是tryToBecomeLeader函数的主要职责，该函数直接使用了redsync包的算法，并利用与redis node建立的连接(NewClient)，尝试建立并持有分布式锁“the-year-of-the-ox-2021”。我们使用的是默认的锁属性，从redsync包的NewMutex方法源码，我们能看到锁默认属性如下：

    1. // github.com/go-redsync/redsync/redsync.go
    2.  
    3. // NewMutex returns a new distributed mutex with given name.
    4. func (r *Redsync) NewMutex(name string, options ...Option) *Mutex {
    5.         m := &Mutex{
    6.                 name:         name,
    7.                 expiry:       8 * time.Second,
    8.                 tries:        32,
    9.                 delayFunc:    func(tries int) time.Duration { return 500 * time.Millisecond },
    10.                 genValueFunc: genValue,
    11.                 factor:       0.01,
    12.                 quorum:       len(r.pools)/2 + 1,
    13.                 pools:        r.pools,
    14.         }
    15.         for _, o := range options {
    16.                 o.Apply(m)
    17.         }
    18.         return m
    19. }

    我们看到锁有一个过期时间属性(expiry)，过期时间默认仅有8秒。问题来了：一旦锁过期了，那么情况会怎样？事实是一旦锁过期掉，在leader尚未解锁时，其follower也会加锁成功，因为原锁的key已经因过期而被删除掉了。长此以往，整个分布式系统就会存在多个自视为leader的进程，整个处理逻辑就乱了！

    解决这个问题至少可以有三种方案：

    • 方案1：将锁的expiry设置得很长，长到一旦某个服务持有了锁，不需担心锁过期的问题；
    • 方案2：在所的默认expiry到期之前解锁，所有服务重新竞争锁；
    • 方案3：一旦某个服务持有了锁，则需要定期重设锁的expiry时间，保证锁不会过期，直到该服务主动执行unlock。

    方案1的问题在于，一旦持有锁的leader因意外异常退出并且尚未unlock，那么由于锁的过期时间超级长，其他follower依然无法持有锁而变成下一任leader，导致整个分布式系统的leader缺失，业务逻辑无法继续进行；

    方案2其实是基于Redis分布式锁的常规使用方式，但对于像我这里的业务场景，频繁lock和unlock没必要，我只需要保证系统中有一个leader一直在处理过期event即可，在服务间轮流处理并非我的需求。但这个方案是一个可行的方案，代码逻辑清晰也简单。

    方案3则是非常适合我的业务场景的方案，持有锁的leader通过定期(<8s)的更新锁的过期时间来保证锁的有效性，这样避免了leader频繁切换。这里我们就使用了这一方案，见78~82行，我们在定时器的帮助下，定期重新设置了锁的过期时间(8s)。

    在上述示例代码中，我们用一个变量isLeader来标识该服务是否持有了锁，由于该变量被多个goroutine访问和修改，因此我们通过atomic包实现对其的原子访问以避免出现race问题。

    最后，我们说说这段示例承载的业务逻辑(doExpire函数)。真正的业务逻辑由doExpire函数实现。它通过监听redis 0号库的key空间的过期事件实现对目标key的过期处理(这里并未体现这一点)。

    subscribe的subject字符串为 keyevent@0 :expired ，这个字符串的组成含义可以参考 redis官方对notifications的说明 ，这里的字串表明我们要监听key事件，在0号数据库，事件类型是key过期。

    当在0号数据库有key过期后，我们的订阅channel(105行)就会收到一个事件，通过event的Payload我们可以得到key的名称，后续我们可以根据key的名字来过滤掉我们不关心的key，而仅对期望的key做相应处理。

    在默认配置下， redis的通知功能处于关闭状态。我们需要通过命令或在redis.conf中开启这一功能。

    1. $redis-cli
    2. 127.0.0.1:6379> config set notify-keyspace-events KEx
    3. OK

    到这里，我们已经搞清楚了上面示例代码的原理，下面我们就来真实运行一次上面的代码，我们编译上面代码并启动三个实例：

    1. $go build main.go
    2. $./main 1
    3. $./main 2
    4. $./main 3

    由于 ./main 1 先启动，因此第一个启动的服务一般会先成为leader：

    1. $main 1
    2. 2021/02/11 05:43:15 prog-1 subscribe expire event ok
    3. 2021/02/11 05:43:20 prog-1 become leader successfully
    4. 2021/02/11 05:43:25 prog-1 is the leader
    5. 2021/02/11 05:43:30 prog-1 is the leader

    而其他两个服务会定期尝试去持有锁：

    1. $main 2
    2. 2021/02/11 05:43:17 prog-2 subscribe expire event ok
    3. 2021/02/11 05:43:37 prog-2 try to become leader failed: redsync: failed to acquire lock
    4. 2021/02/11 05:43:53 prog-2 try to become leader failed: redsync: failed to acquire lock
    5.  
    6. $main 3
    7. 2021/02/11 05:43:18 prog-3 subscribe expire event ok
    8. 2021/02/11 05:43:38 prog-3 try to become leader failed: redsync: failed to acquire lock
    9. 2021/02/11 05:43:54 prog-3 try to become leader failed: redsync: failed to acquire lock

    这时我们通过redis-cli在0号数据库中创建一个key1，过期时间5s：

    1. $redis-cli
    2. 127.0.0.1:6379> setex key1 5 value1
    3. OK

    5s后，我们会在prog-1这个服务实例的输出日志中看到如下内容：

    1. 2021/02/11 05:43:50 prog-1 is the leader
    2. 2021/02/11 05:43:53 prog-1 收到并处理一条过期消息[key:key1]
    3. 2021/02/11 05:43:55 prog-1 is the leader

    接下来，我们停掉prog-1：

    1. 2021/02/11 05:44:00 prog-1 is the leader
    2. ^C2021/02/11 05:44:01 recv exit signal...
    3. redis: 2021/02/11 05:44:01 pubsub.go:168: redis: discarding bad PubSub connection: read tcp [::1]:56594->[::1]:6379: use of closed network connection
    4. 2021/02/11 05:44:01 program exit ok

    在停掉prog-1后的瞬间，prog-2成功持有了锁，并成为leader：

    1. 2021/02/11 05:44:01 prog-2 become leader successfully
    2. 2021/02/11 05:44:01 prog-2 is the leader

    我们再通过redis-cli在0号数据库中创建一个key2，过期时间5s：

    1. $redis-cli
    2. 127.0.0.1:6379> setex key2 5 value2
    3. OK

    5s后，我们会在prog-2这个服务实例的输出日志中看到如下内容：

    1. 2021/02/11 05:44:17 prog-2 is the leader
    2. 2021/02/11 05:44:19 prog-2 收到并处理一条过期消息[key:key2]
    3. 2021/02/11 05:44:22 prog-2 is the leader

    从运行的结果来看，该分布式系统的运行逻辑是符合我们的设计预期的。

    2. 基于redis集群的分布式锁

    上面，我们实现了基于单个redis节点的分布式锁的选主功能。在生产环境，我们很少会使用单节点的Redis，通常会使用Redis集群以保证高可用性。

    最新的 redsync已经支持了redis cluster(基于go-redis) 。和单节点唯一不同的是，我们传递给redsync的pool所使用的与redis的连接由Client类型变为了ClusterClient类型：

    1. // github.com/bigwhite/experiments/blob/master/redis-cluster-distributed-lock/cluster/v1/main.go
    2. const (
    3.         redisClusterMasters      = "localhost:30001,localhost:30002,localhost:30003"
    4. )
    5.  
    6. func main() {
    7.     ... ...
    8.         client := goredislib.NewClusterClient(&goredislib.ClusterOptions{
    9.                 Addrs: strings.Split(redisClusterMasters, ",")})
    10.         defer client.Close()
    11.     ... ...
    12. }

    我们在本地启动的redis cluster，三个master的地址分别为：localhost:30001、localhost:30002和localhost:30003。我们将master的地址组成一个逗号分隔的常量redisClusterMasters。

    我们对上面单节点的代码做了改进，将Redis连接的创建放在了main中，并将client连接作为参数传递给各个goroutine的运行函数。下面是cluster版示例代码完整版(v1)：

    1. // github.com/bigwhite/experiments/blob/master/redis-cluster-distributed-lock/cluster/v1/main.go
    2.  
    3.      1  package main
    4.      2
    5.      3  import (
    6.      4      "context"
    7.      5      "log"
    8.      6      "os"
    9.      7      "os/signal"
    10.      8      "strings"
    11.      9      "sync"
    12.     10      "sync/atomic"
    13.     11      "syscall"
    14.     12      "time"
    15.     13
    16.     14      goredislib "github.com/go-redis/redis/v8"
    17.     15      "github.com/go-redsync/redsync/v4"
    18.     16      "github.com/go-redsync/redsync/v4/redis/goredis/v8"
    19.     17  )
    20.     18
    21.     19  const (
    22.     20      redisKeyExpiredEventSubj = `__keyevent@0__:expired`
    23.     21      redisClusterMasters      = "localhost:30001,localhost:30002,localhost:30003"
    24.     22  )
    25.     23
    26.     24  var (
    27.     25      isLeader  int64
    28.     26      m         atomic.Value
    29.     27      id        string
    30.     28      mutexName = "the-year-of-the-ox-2021"
    31.     29  )
    32.     30
    33.     31  func init() {
    34.     32      if len(os.Args) < 2 {
    35.     33          panic("args number is not correct")
    36.     34      }
    37.     35      id = os.Args[1]
    38.     36  }
    39.     37
    40.     38  func tryToBecomeLeader(client *goredislib.ClusterClient) (bool, func() (bool, error), error) {
    41.     39      pool := goredis.NewPool(client)
    42.     40      rs := redsync.New(pool)
    43.     41
    44.     42      mutex := rs.NewMutex(mutexName)
    45.     43
    46.     44      if err := mutex.Lock(); err != nil {
    47.     45          return false, nil, err
    48.     46      }
    49.     47
    50.     48      return true, func() (bool, error) {
    51.     49          return mutex.Unlock()
    52.     50      }, nil
    53.     51  }
    54.     52
    55.     53  func doElectionAndMaintainTheStatus(c *goredislib.ClusterClient, quit <-chan struct{}) {
    56.     54      ticker := time.NewTicker(time.Second * 5)
    57.     55      var err error
    58.     56      var ok bool
    59.     57      var cf func() (bool, error)
    60.     58
    61.     59      for {
    62.     60          select {
    63.     61          case <-ticker.C:
    64.     62              if atomic.LoadInt64(&isLeader) == 0 {
    65.     63                  ok, cf, err = tryToBecomeLeader(c)
    66.     64                  if ok {
    67.     65                      log.Printf("prog-%s become leader successfully\n", id)
    68.     66                      atomic.StoreInt64(&isLeader, 1)
    69.     67                      defer cf()
    70.     68                  }
    71.     69                  if !ok || err != nil {
    72.     70                      log.Printf("prog-%s try to become leader failed: %s\n", id, err)
    73.     71                  }
    74.     72              } else {
    75.     73                  log.Printf("prog-%s is the leader\n", id)
    76.     74                  // update the lock live time and maintain the leader status
    77.     75                  c.Expire(context.Background(), mutexName, 8*time.Second)
    78.     76              }
    79.     77          case <-quit:
    80.     78              return
    81.     79          }
    82.     80      }
    83.     81  }
    84.     82
    85.     83  func doExpire(c *goredislib.ClusterClient, quit <-chan struct{}) {
    86.     84      // subscribe the expire event of redis
    87.     85      ctx := context.Background()
    88.     86      pubsub := c.Subscribe(ctx, redisKeyExpiredEventSubj)
    89.     87      _, err := pubsub.Receive(ctx)
    90.     88      if err != nil {
    91.     89          log.Printf("prog-%s subscribe expire event failed: %s\n", id, err)
    92.     90          return
    93.     91      }
    94.     92      log.Printf("prog-%s subscribe expire event ok\n", id)
    95.     93
    96.     94      // Go channel which receives messages from redis db
    97.     95      ch := pubsub.Channel()
    98.     96      for {
    99.     97          select {
    100.     98          case event := <-ch:
    101.     99              key := event.Payload
    102.    100              if atomic.LoadInt64(&isLeader) == 0 {
    103.    101                  break
    104.    102              }
    105.    103              log.Printf("prog-%s 收到并处理一条过期消息[key:%s]", id, key)
    106.    104          case <-quit:
    107.    105              return
    108.    106          }
    109.    107      }
    110.    108  }
    111.    109
    112.    110  func main() {
    113.    111      var wg sync.WaitGroup
    114.    112      wg.Add(2)
    115.    113      var quit = make(chan struct{})
    116.    114      client := goredislib.NewClusterClient(&goredislib.ClusterOptions{
    117.    115          Addrs: strings.Split(redisClusterMasters, ",")})
    118.    116      defer client.Close()
    119.    117
    120.    118      go func() {
    121.    119          doElectionAndMaintainTheStatus(client, quit)
    122.    120          wg.Done()
    123.    121      }()
    124.    122      go func() {
    125.    123          doExpire(client, quit)
    126.    124          wg.Done()
    127.    125      }()
    128.    126
    129.    127      c := make(chan os.Signal, 1)
    130.    128      signal.Notify(c, syscall.SIGINT, syscall.SIGTERM)
    131.    129      _ = <-c
    132.    130      close(quit)
    133.    131      log.Printf("recv exit signal...")
    134.    132      wg.Wait()
    135.    133      log.Printf("program exit ok")
    136.    134  }

    和单一节点一样，我们运行三个服务实例：

    1. $go build main.go
    2. $main 1
    3. 2021/02/11 09:49:16 prog-1 subscribe expire event ok
    4. 2021/02/11 09:49:22 prog-1 become leader successfully
    5. 2021/02/11 09:49:26 prog-1 is the leader
    6. 2021/02/11 09:49:31 prog-1 is the leader
    7. 2021/02/11 09:49:36 prog-1 is the leader
    8. ... ...
    9.  
    10. $main 2
    11. 2021/02/11 09:49:19 prog-2 subscribe expire event ok
    12. 2021/02/11 09:49:40 prog-2 try to become leader failed: redsync: failed to acquire lock
    13. 2021/02/11 09:49:55 prog-2 try to become leader failed: redsync: failed to acquire lock
    14. ... ...
    15.  
    16. $main 3
    17. 2021/02/11 09:49:31 prog-3 subscribe expire event ok
    18. 2021/02/11 09:49:52 prog-3 try to become leader failed: redsync: failed to acquire lock
    19. 2021/02/11 09:50:07 prog-3 try to become leader failed: redsync: failed to acquire lock
    20. ... ...

    我们看到基于Redis集群版的分布式锁也生效了！prog-1成功持有锁并成为leader! 接下来我们再来看看对过期key事件的处理！

    我们通过下面命令让redis-cli连接到集群中的所有节点并设置每个节点开启key空间的事件通知：

    1. 三主：
    2.  
    3. $redis-cli -c -h localhost -p 30001
    4. localhost:30001> config set notify-keyspace-events KEx
    5. OK
    6.  
    7. $redis-cli -c -h localhost -p 30002
    8. localhost:30002> config set notify-keyspace-events KEx
    9. OK
    10.  
    11. $redis-cli -c -h localhost -p 30003
    12. localhost:30003> config set notify-keyspace-events KEx
    13. OK
    14.  
    15. 三从：
    16.  
    17. $redis-cli -c -h localhost -p 30004
    18. localhost:30004> config set notify-keyspace-events KEx
    19. OK
    20.  
    21. $redis-cli -c -h localhost -p 30005
    22. localhost:30005> config set notify-keyspace-events KEx
    23. OK
    24.  
    25. $redis-cli -c -h localhost -p 30006
    26. localhost:30006> config set notify-keyspace-events KEx
    27. OK

    在node1节点上，我们set一个有效期为5s的key：key1：

    1. localhost:30001> setex key1 5 value1
    2. -> Redirected to slot [9189] located at 127.0.0.1:30002
    3. OK

    等待5s后，我们的leader：prog-1 并没有如预期那样受到expire通知！ 这是怎么回事呢？追本溯源，我们查看一下 redis官方文档关于notifications的说明 ，我们在文档最后一段找到如下描述：

    1. Events in a cluster
    2.  
    3. Every node of a Redis cluster generates events about its own subset of the keyspace as described above. However, unlike regular Pub/Sub communication in a cluster, events' notifications are not broadcasted to all nodes. Put differently, keyspace events are node-specific. This means that to receive all keyspace events of a cluster, clients need to subscribe to each of the nodes.

    这段话大致意思是Redis集群中的每个redis node都有自己的keyspace，事件通知不会被广播到集群内的所有节点，即keyspace的事件是node相关的。如果要接收一个集群中的所有keyspace的event，那客户端就需要Subcribe集群内的所有节点。我们来改一下代码，形成v2版(考虑到篇幅就不列出所有代码了，仅列出相对于v1版变化的代码)：

    1. // github.com/bigwhite/experiments/blob/master/redis-cluster-distributed-lock/cluster/v2/main.go
    2.  
    3. ... ...
    4.     19  const (
    5.     20      redisKeyExpiredEventSubj = `__keyevent@0__:expired`
    6.     21      redisClusterMasters      = "localhost:30001,localhost:30002,localhost:30003,localhost:30004,localhost:30005,localhost:30006"
    7.     22  )
    8. ... ...
    9.     83  func doExpire(quit <-chan struct{}) {
    10.     84      var ch = make(chan *goredislib.Message)
    11.     85      nodes := strings.Split(redisClusterMasters, ",")
    12.     86
    13.     87      for _, node := range nodes {
    14.     88          node := node
    15.     89          go func(quit <-chan struct{}) {
    16.     90              c := goredislib.NewClient(&goredislib.Options{
    17.     91                  Addr: node})
    18.     92              defer c.Close()
    19.     93
    20.     94              // subscribe the expire event of redis
    21.     95              ctx := context.Background()
    22.     96              pubsub := c.Subscribe(ctx, redisKeyExpiredEventSubj)
    23.     97              _, err := pubsub.Receive(ctx)
    24.     98              if err != nil {
    25.     99                  log.Printf("prog-%s subscribe expire event of node[%s] failed: %s\n",
    26.    100                      id, node, err)
    27.    101                  return
    28.    102              }
    29.    103              log.Printf("prog-%s subscribe expire event of node[%s] ok\n", id, node)
    30.    104
    31.    105              // Go channel which receives messages from redis db
    32.    106              pch := pubsub.Channel()
    33.    107
    34.    108              for {
    35.    109                  select {
    36.    110                  case event := <-pch:
    37.    111                      ch <- event
    38.    112                  case <-quit:
    39.    113                      return
    40.    114                  }
    41.    115              }
    42.    116          }(quit)
    43.    117      }
    44.    118      for {
    45.    119          select {
    46.    120          case event := <-ch:
    47.    121              key := event.Payload
    48.    122              if atomic.LoadInt64(&isLeader) == 0 {
    49.    123                  break
    50.    124              }
    51.    125              log.Printf("prog-%s 收到并处理一条过期消息[key:%s]", id, key)
    52.    126          case <-quit:
    53.    127              return
    54.    128          }
    55.    129      }
    56.    130  }
    57.    131
    58.    132  func main() {
    59.    133      var wg sync.WaitGroup
    60.    134      wg.Add(2)
    61.    135      var quit = make(chan struct{})
    62.    136      client := goredislib.NewClusterClient(&goredislib.ClusterOptions{
    63.    137          Addrs: strings.Split(redisClusterMasters, ",")})
    64.    138      defer client.Close()
    65.    139
    66.    140      go func() {
    67.    141          doElectionAndMaintainTheStatus(client, quit)
    68.    142          wg.Done()
    69.    143      }()
    70.    144      go func() {
    71.    145          doExpire(quit)
    72.    146          wg.Done()
    73.    147      }()
    74.    148
    75.    149      c := make(chan os.Signal, 1)
    76.    150      signal.Notify(c, syscall.SIGINT, syscall.SIGTERM)
    77.    151      _ = <-c
    78.    152      close(quit)
    79.    153      log.Printf("recv exit signal...")
    80.    154      wg.Wait()
    81.    155      log.Printf("program exit ok")
    82.    156  }

    在这个新版代码中，我们在每个新goroutine中实现对redis一个节点的Subscribe，并将收到的Event notifications通过“扇入”模式(更多关于并发扇入模式的内容，可以参考我的Go技术专栏文章 《Go并发模型和常见并发模式》 )统一写入到运行doExpire的goroutine中做统一处理。

    我们再来运行一下这个示例，并在不同时机创建多个key来验证通知接收和处理的效果：

    1. $main 1
    2. 2021/02/11 10:29:21 prog-1 subscribe expire event of node[localhost:30004] ok
    3. 2021/02/11 10:29:21 prog-1 subscribe expire event of node[localhost:30001] ok
    4. 2021/02/11 10:29:21 prog-1 subscribe expire event of node[localhost:30006] ok
    5. 2021/02/11 10:29:21 prog-1 subscribe expire event of node[localhost:30002] ok
    6. 2021/02/11 10:29:21 prog-1 subscribe expire event of node[localhost:30003] ok
    7. 2021/02/11 10:29:21 prog-1 subscribe expire event of node[localhost:30005] ok
    8. 2021/02/11 10:29:26 prog-1 become leader successfully
    9. 2021/02/11 10:29:31 prog-1 is the leader
    10. 2021/02/11 10:29:36 prog-1 is the leader
    11. 2021/02/11 10:29:41 prog-1 is the leader
    12. 2021/02/11 10:29:46 prog-1 is the leader
    13. 2021/02/11 10:29:47 prog-1 收到并处理一条过期消息[key:key1]
    14. 2021/02/11 10:29:51 prog-1 is the leader
    15. 2021/02/11 10:29:51 prog-1 收到并处理一条过期消息[key:key2]
    16. 2021/02/11 10:29:56 prog-1 收到并处理一条过期消息[key:key3]
    17. 2021/02/11 10:29:56 prog-1 is the leader
    18. 2021/02/11 10:30:01 prog-1 is the leader
    19. 2021/02/11 10:30:06 prog-1 is the leader
    20. ^C2021/02/11 10:30:08 recv exit signal...
    21.  
    22. $main 3
    23. 2021/02/11 10:29:27 prog-3 subscribe expire event of node[localhost:30004] ok
    24. 2021/02/11 10:29:27 prog-3 subscribe expire event of node[localhost:30006] ok
    25. 2021/02/11 10:29:27 prog-3 subscribe expire event of node[localhost:30002] ok
    26. 2021/02/11 10:29:27 prog-3 subscribe expire event of node[localhost:30001] ok
    27. 2021/02/11 10:29:27 prog-3 subscribe expire event of node[localhost:30005] ok
    28. 2021/02/11 10:29:27 prog-3 subscribe expire event of node[localhost:30003] ok
    29. 2021/02/11 10:29:48 prog-3 try to become leader failed: redsync: failed to acquire lock
    30. 2021/02/11 10:30:03 prog-3 try to become leader failed: redsync: failed to acquire lock
    31. 2021/02/11 10:30:08 prog-3 become leader successfully
    32. 2021/02/11 10:30:08 prog-3 is the leader
    33. 2021/02/11 10:30:12 prog-3 is the leader
    34. 2021/02/11 10:30:17 prog-3 is the leader
    35. 2021/02/11 10:30:22 prog-3 is the leader
    36. 2021/02/11 10:30:23 prog-3 收到并处理一条过期消息[key:key4]
    37. 2021/02/11 10:30:27 prog-3 is the leader
    38. ^C2021/02/11 10:30:28 recv exit signal...
    39.  
    40. $main 2
    41. 2021/02/11 10:29:24 prog-2 subscribe expire event of node[localhost:30005] ok
    42. 2021/02/11 10:29:24 prog-2 subscribe expire event of node[localhost:30006] ok
    43. 2021/02/11 10:29:24 prog-2 subscribe expire event of node[localhost:30003] ok
    44. 2021/02/11 10:29:24 prog-2 subscribe expire event of node[localhost:30004] ok
    45. 2021/02/11 10:29:24 prog-2 subscribe expire event of node[localhost:30002] ok
    46. 2021/02/11 10:29:24 prog-2 subscribe expire event of node[localhost:30001] ok
    47. 2021/02/11 10:29:45 prog-2 try to become leader failed: redsync: failed to acquire lock
    48. 2021/02/11 10:30:01 prog-2 try to become leader failed: redsync: failed to acquire lock
    49. 2021/02/11 10:30:16 prog-2 try to become leader failed: redsync: failed to acquire lock
    50. 2021/02/11 10:30:28 prog-2 become leader successfully
    51. 2021/02/11 10:30:28 prog-2 is the leader
    52. 2021/02/11 10:30:29 prog-2 is the leader
    53. 2021/02/11 10:30:34 prog-2 is the leader
    54. 2021/02/11 10:30:39 prog-2 收到并处理一条过期消息[key:key5]
    55. 2021/02/11 10:30:39 prog-2 is the leader
    56. ^C2021/02/11 10:30:41 recv exit signal...

    这个运行结果如预期！

    不过这个方案显然也不是那么理想，毕竟我们要单独Subscribe每个集群内的redis节点，目前没有理想方案，除非redis cluster支持带广播的Event notification

  • 相关阅读:
    Netty入门——概述
    代码随想录Day42-图论：力扣第417m、841m、463e题
    成都瀚网科技有限公司抖音带货正规么
    UDP文件传输工具之UDP怎么限流
    headscale的部署方法和使用教程
    const关键字用法总结
    国华小状元1号年金险怎么样？好不好？
    vue3 使用 mitt 插件实现非父子组件传值
    swift-类结构源码探寻(二)
    墨水屏技术在贴片厂的创新应用探索
  • 原文地址：https://blog.csdn.net/Q54665642ljf/article/details/126196318