标签 选主 下的文章

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

img{512x368}

今天要说的技术方案也是有一定项目背景的。在上一个项目中,我们需要对一个redis集群中过期的key进行处理,这是一个分布式
系统,考虑到高可用性,需要具备过期处理功能的服务有多个副本,这样我们就要求在同一时间内仅有一个副本可以对过期的key>进行处理,如果该副本挂掉,系统会在其他副本中再挑选出一个来处理过期的key。

很显然,这里涉及到一个选主(leader election)的过程。每当涉及选主,很多人就会想到一些高大上的分布式一致性/共识算法,
比如:raftpaxos等。当然使用这
些算法自然没有问题,但是也给系统徒增了很多复杂性。能否有一些更简单直接的方案呢?我们已经有了一个redis集群,是否可>以利用redis集群的能力来完成这一点呢?

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

img{512x368}

图:利用redis分布式锁实现选主

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

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

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

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

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

// github.com/bigwhite/experiments/blob/master/redis-cluster-distributed-lock/standalone/main.go

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

// github.com/go-redsync/redsync/redsync.go

// NewMutex returns a new distributed mutex with given name.
func (r *Redsync) NewMutex(name string, options ...Option) *Mutex {
        m := &Mutex{
                name:         name,
                expiry:       8 * time.Second,
                tries:        32,
                delayFunc:    func(tries int) time.Duration { return 500 * time.Millisecond },
                genValueFunc: genValue,
                factor:       0.01,
                quorum:       len(r.pools)/2 + 1,
                pools:        r.pools,
        }
        for _, o := range options {
                o.Apply(m)
        }
        return m
}

我们看到锁有一个过期时间属性(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中开启这一功能。

$redis-cli
127.0.0.1:6379> config set notify-keyspace-events KEx
OK

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

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

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

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

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

$main 2
2021/02/11 05:43:17 prog-2 subscribe expire event ok
2021/02/11 05:43:37 prog-2 try to become leader failed: redsync: failed to acquire lock
2021/02/11 05:43:53 prog-2 try to become leader failed: redsync: failed to acquire lock

$main 3
2021/02/11 05:43:18 prog-3 subscribe expire event ok
2021/02/11 05:43:38 prog-3 try to become leader failed: redsync: failed to acquire lock
2021/02/11 05:43:54 prog-3 try to become leader failed: redsync: failed to acquire lock

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

$redis-cli
127.0.0.1:6379> setex key1 5 value1
OK

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

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

接下来,我们停掉prog-1:

2021/02/11 05:44:00 prog-1 is the leader
^C2021/02/11 05:44:01 recv exit signal...
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
2021/02/11 05:44:01 program exit ok

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

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

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

$redis-cli
127.0.0.1:6379> setex key2 5 value2
OK

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

2021/02/11 05:44:17 prog-2 is the leader
2021/02/11 05:44:19 prog-2 收到并处理一条过期消息[key:key2]
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类型:

// github.com/bigwhite/experiments/blob/master/redis-cluster-distributed-lock/cluster/v1/main.go
const (
        redisClusterMasters      = "localhost:30001,localhost:30002,localhost:30003"
)

func main() {
    ... ...
        client := goredislib.NewClusterClient(&goredislib.ClusterOptions{
                Addrs: strings.Split(redisClusterMasters, ",")})
        defer client.Close()
    ... ...
}

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

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

// github.com/bigwhite/experiments/blob/master/redis-cluster-distributed-lock/cluster/v1/main.go

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

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

$go build main.go
$main 1
2021/02/11 09:49:16 prog-1 subscribe expire event ok
2021/02/11 09:49:22 prog-1 become leader successfully
2021/02/11 09:49:26 prog-1 is the leader
2021/02/11 09:49:31 prog-1 is the leader
2021/02/11 09:49:36 prog-1 is the leader
... ...

$main 2
2021/02/11 09:49:19 prog-2 subscribe expire event ok
2021/02/11 09:49:40 prog-2 try to become leader failed: redsync: failed to acquire lock
2021/02/11 09:49:55 prog-2 try to become leader failed: redsync: failed to acquire lock
... ...

$main 3
2021/02/11 09:49:31 prog-3 subscribe expire event ok
2021/02/11 09:49:52 prog-3 try to become leader failed: redsync: failed to acquire lock
2021/02/11 09:50:07 prog-3 try to become leader failed: redsync: failed to acquire lock
... ...

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

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

三主:

$redis-cli -c -h localhost -p 30001
localhost:30001> config set notify-keyspace-events KEx
OK

$redis-cli -c -h localhost -p 30002
localhost:30002> config set notify-keyspace-events KEx
OK

$redis-cli -c -h localhost -p 30003
localhost:30003> config set notify-keyspace-events KEx
OK

三从:

$redis-cli -c -h localhost -p 30004
localhost:30004> config set notify-keyspace-events KEx
OK

$redis-cli -c -h localhost -p 30005
localhost:30005> config set notify-keyspace-events KEx
OK

$redis-cli -c -h localhost -p 30006
localhost:30006> config set notify-keyspace-events KEx
OK

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

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

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

Events in a cluster

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版变化的代码):

// github.com/bigwhite/experiments/blob/master/redis-cluster-distributed-lock/cluster/v2/main.go

... ...
    19  const (
    20      redisKeyExpiredEventSubj = `__keyevent@0__:expired`
    21      redisClusterMasters      = "localhost:30001,localhost:30002,localhost:30003,localhost:30004,localhost:30005,localhost:30006"
    22  )
... ...
    83  func doExpire(quit <-chan struct{}) {
    84      var ch = make(chan *goredislib.Message)
    85      nodes := strings.Split(redisClusterMasters, ",")
    86
    87      for _, node := range nodes {
    88          node := node
    89          go func(quit <-chan struct{}) {
    90              c := goredislib.NewClient(&goredislib.Options{
    91                  Addr: node})
    92              defer c.Close()
    93
    94              // subscribe the expire event of redis
    95              ctx := context.Background()
    96              pubsub := c.Subscribe(ctx, redisKeyExpiredEventSubj)
    97              _, err := pubsub.Receive(ctx)
    98              if err != nil {
    99                  log.Printf("prog-%s subscribe expire event of node[%s] failed: %s\n",
   100                      id, node, err)
   101                  return
   102              }
   103              log.Printf("prog-%s subscribe expire event of node[%s] ok\n", id, node)
   104
   105              // Go channel which receives messages from redis db
   106              pch := pubsub.Channel()
   107
   108              for {
   109                  select {
   110                  case event := <-pch:
   111                      ch <- event
   112                  case <-quit:
   113                      return
   114                  }
   115              }
   116          }(quit)
   117      }
   118      for {
   119          select {
   120          case event := <-ch:
   121              key := event.Payload
   122              if atomic.LoadInt64(&isLeader) == 0 {
   123                  break
   124              }
   125              log.Printf("prog-%s 收到并处理一条过期消息[key:%s]", id, key)
   126          case <-quit:
   127              return
   128          }
   129      }
   130  }
   131
   132  func main() {
   133      var wg sync.WaitGroup
   134      wg.Add(2)
   135      var quit = make(chan struct{})
   136      client := goredislib.NewClusterClient(&goredislib.ClusterOptions{
   137          Addrs: strings.Split(redisClusterMasters, ",")})
   138      defer client.Close()
   139
   140      go func() {
   141          doElectionAndMaintainTheStatus(client, quit)
   142          wg.Done()
   143      }()
   144      go func() {
   145          doExpire(quit)
   146          wg.Done()
   147      }()
   148
   149      c := make(chan os.Signal, 1)
   150      signal.Notify(c, syscall.SIGINT, syscall.SIGTERM)
   151      _ = <-c
   152      close(quit)
   153      log.Printf("recv exit signal...")
   154      wg.Wait()
   155      log.Printf("program exit ok")
   156  }

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

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

$main 1
2021/02/11 10:29:21 prog-1 subscribe expire event of node[localhost:30004] ok
2021/02/11 10:29:21 prog-1 subscribe expire event of node[localhost:30001] ok
2021/02/11 10:29:21 prog-1 subscribe expire event of node[localhost:30006] ok
2021/02/11 10:29:21 prog-1 subscribe expire event of node[localhost:30002] ok
2021/02/11 10:29:21 prog-1 subscribe expire event of node[localhost:30003] ok
2021/02/11 10:29:21 prog-1 subscribe expire event of node[localhost:30005] ok
2021/02/11 10:29:26 prog-1 become leader successfully
2021/02/11 10:29:31 prog-1 is the leader
2021/02/11 10:29:36 prog-1 is the leader
2021/02/11 10:29:41 prog-1 is the leader
2021/02/11 10:29:46 prog-1 is the leader
2021/02/11 10:29:47 prog-1 收到并处理一条过期消息[key:key1]
2021/02/11 10:29:51 prog-1 is the leader
2021/02/11 10:29:51 prog-1 收到并处理一条过期消息[key:key2]
2021/02/11 10:29:56 prog-1 收到并处理一条过期消息[key:key3]
2021/02/11 10:29:56 prog-1 is the leader
2021/02/11 10:30:01 prog-1 is the leader
2021/02/11 10:30:06 prog-1 is the leader
^C2021/02/11 10:30:08 recv exit signal...

$main 3
2021/02/11 10:29:27 prog-3 subscribe expire event of node[localhost:30004] ok
2021/02/11 10:29:27 prog-3 subscribe expire event of node[localhost:30006] ok
2021/02/11 10:29:27 prog-3 subscribe expire event of node[localhost:30002] ok
2021/02/11 10:29:27 prog-3 subscribe expire event of node[localhost:30001] ok
2021/02/11 10:29:27 prog-3 subscribe expire event of node[localhost:30005] ok
2021/02/11 10:29:27 prog-3 subscribe expire event of node[localhost:30003] ok
2021/02/11 10:29:48 prog-3 try to become leader failed: redsync: failed to acquire lock
2021/02/11 10:30:03 prog-3 try to become leader failed: redsync: failed to acquire lock
2021/02/11 10:30:08 prog-3 become leader successfully
2021/02/11 10:30:08 prog-3 is the leader
2021/02/11 10:30:12 prog-3 is the leader
2021/02/11 10:30:17 prog-3 is the leader
2021/02/11 10:30:22 prog-3 is the leader
2021/02/11 10:30:23 prog-3 收到并处理一条过期消息[key:key4]
2021/02/11 10:30:27 prog-3 is the leader
^C2021/02/11 10:30:28 recv exit signal...

$main 2
2021/02/11 10:29:24 prog-2 subscribe expire event of node[localhost:30005] ok
2021/02/11 10:29:24 prog-2 subscribe expire event of node[localhost:30006] ok
2021/02/11 10:29:24 prog-2 subscribe expire event of node[localhost:30003] ok
2021/02/11 10:29:24 prog-2 subscribe expire event of node[localhost:30004] ok
2021/02/11 10:29:24 prog-2 subscribe expire event of node[localhost:30002] ok
2021/02/11 10:29:24 prog-2 subscribe expire event of node[localhost:30001] ok
2021/02/11 10:29:45 prog-2 try to become leader failed: redsync: failed to acquire lock
2021/02/11 10:30:01 prog-2 try to become leader failed: redsync: failed to acquire lock
2021/02/11 10:30:16 prog-2 try to become leader failed: redsync: failed to acquire lock
2021/02/11 10:30:28 prog-2 become leader successfully
2021/02/11 10:30:28 prog-2 is the leader
2021/02/11 10:30:29 prog-2 is the leader
2021/02/11 10:30:34 prog-2 is the leader
2021/02/11 10:30:39 prog-2 收到并处理一条过期消息[key:key5]
2021/02/11 10:30:39 prog-2 is the leader
^C2021/02/11 10:30:41 recv exit signal...

这个运行结果如预期!

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

以上示例代码可以在这里 https://github.com/bigwhite/experiments/tree/master/redis-cluster-distributed-lock 下载 。


“Gopher部落”知识星球开球了!高品质首发Go技术文章,“三天”首发阅读权,每年两期Go语言发展现状分析,每天提前1小时阅读到新鲜的Gopher日报,网课、技术专栏、图书内容前瞻,六小时内必答保证等满足你关于Go语言生态的所有需求!星球首开,福利自然是少不了的!2020年年底之前,8.8折(很吉利吧^_^)加入星球,下方图片扫起来吧!

Go技术专栏“改善Go语⾔编程质量的50个有效实践”正在慕课网火热热销中!本专栏主要满足广大gopher关于Go语言进阶的需求,围绕如何写出地道且高质量Go代码给出50条有效实践建议,上线后收到一致好评!欢迎大家订阅!

我的网课“Kubernetes实战:高可用集群搭建、配置、运维与应用”在慕课网热卖中,欢迎小伙伴们订阅学习!

img{512x368}

我爱发短信:企业级短信平台定制开发专家 https://tonybai.com/。smspush : 可部署在企业内部的定制化短信平台,三网覆盖,不惧大并发接入,可定制扩展; 短信内容你来定,不再受约束, 接口丰富,支持长短信,签名可选。2020年4月8日,中国三大电信运营商联合发布《5G消息白皮书》,51短信平台也会全新升级到“51商用消息平台”,全面支持5G RCS消息。

著名云主机服务厂商DigitalOcean发布最新的主机计划,入门级Droplet配置升级为:1 core CPU、1G内存、25G高速SSD,价格5$/月。有使用DigitalOcean需求的朋友,可以打开这个链接地址:https://m.do.co/c/bff6eed92687 开启你的DO主机之路。

Gopher Daily(Gopher每日新闻)归档仓库 – https://github.com/bigwhite/gopherdaily

我的联系方式:

  • 微博:https://weibo.com/bigwhite20xx
  • 微信公众号:iamtonybai
  • 博客:tonybai.com
  • github: https://github.com/bigwhite
  • “Gopher部落”知识星球:https://public.zsxq.com/groups/51284458844544

微信赞赏:
img{512x368}

商务合作方式:撰稿、出书、培训、在线课程、合伙创业、咨询、广告合作。

使用consul实现分布式服务注册和发现

ConsulHashiCorp公司推出的开源工具,用于实现分布式系统的服务发现与配置。与其他分布式服务注册与发现的方案,比如 AirbnbSmartStack等相比,Consul的方案更“一站式”,内置了服务注册与发现框 架、分布一致性协议实现、健康检查、Key/Value存储、多数据中心方案,不再需要依赖其他工具(比如ZooKeeper等)。使用起来也较 为简单。Consul用Golang实现,因此具有天然可移植性(支持Linux、windows和Mac OS X);安装包仅包含一个可执行文件,方便部署,与Docker等轻量级容器可无缝配合

本文是Consul的入门介绍,并用一些例子说明如何使用Consul实现服务的注册和发现。

一、建立Consul Cluster

要想利用Consul提供的服务实现服务的注册与发现,我们需要建立Consul Cluster。在Consul方案中,每个提供服务的节点上都要部署和运行Consul的agent,所有运行Consul agent节点的集合构成Consul Cluster。Consul agent有两种运行模式:Server和Client。这里的Server和Client只是Consul集群层面的区分,与搭建在Cluster之上 的应用服务无关。以Server模式运行的Consul agent节点用于维护Consul集群的状态,官方建议每个Consul Cluster至少有3个或以上的运行在Server mode的Agent,Client节点不限。

每个数据中心的Consul Cluster都会在运行于server模式下的agent节点中选出一个Leader节点,这个选举过程通过Consul实现的raft协议保证,多个 server节点上的Consul数据信息是强一致的。处于client mode的Consul agent节点比较简单,无状态,仅仅负责将请求转发给Server agent节点。

下面我们就来搭建一个实验Consul Cluster。

实验环境和节点角色如下:

n1(Ubuntu 14.04 x86_64): 10.10.105.71  server mode
n2(Ubuntu 12.04 x86_64): 10.10.126.101 server mode    with Consul Web UI
n3(Ubuntu 9.04 i386): 10.10.126.187    client mode

在三台主机上分别下载和安装Consul包,安装包很简单,只是包含一个可执行文件consul。在n2主机上还要下载一份Consul Web UI包,支持图形化展示Consul cluster中的节点状态和服务状态。

Consul Cluster的启动过程如下:

n1主机:

$ consul agent -server -bootstrap-expect 2 -data-dir /tmp/consul -node=n1 -bind=10.10.105.71 -dc=dc1
==> WARNING: Expect Mode enabled, expecting 2 servers
==> WARNING: It is highly recommended to set GOMAXPROCS higher than 1
==> Starting Consul agent…
==> Starting Consul agent RPC…
==> Consul agent running!
         Node name: 'n1'
        Datacenter: 'dc1'
            Server: true (bootstrap: false)
       Client Addr: 127.0.0.1 (HTTP: 8500, HTTPS: -1, DNS: 8600, RPC: 8400)
      Cluster Addr: 10.10.105.71 (LAN: 8301, WAN: 8302)
    Gossip encrypt: false, RPC-TLS: false, TLS-Incoming: false
             Atlas: <disabled>

==> Log data will now stream in as it occurs:

    2015/07/03 09:18:25 [INFO] serf: EventMemberJoin: n1 10.10.105.71
    2015/07/03 09:18:25 [INFO] serf: EventMemberJoin: n1.dc1 10.10.105.71
    2015/07/03 09:18:25 [INFO] raft: Node at 10.10.105.71:8300 [Follower] entering Follower state
    2015/07/03 09:18:25 [INFO] consul: adding server n1 (Addr: 10.10.105.71:8300) (DC: dc1)
    2015/07/03 09:18:25 [INFO] consul: adding server n1.dc1 (Addr: 10.10.105.71:8300) (DC: dc1)
    2015/07/03 09:18:25 [ERR] agent: failed to sync remote state: No cluster leader
    2015/07/03 09:18:26 [WARN] raft: EnableSingleNode disabled, and no known peers. Aborting election.1

n2主机:

$ consul agent -server -bootstrap-expect 2 -data-dir /tmp/consul -node=n2 -bind=10.10.126.101 -ui-dir ./dist  -dc=dc1
==> WARNING: Expect Mode enabled, expecting 2 servers
==> WARNING: It is highly recommended to set GOMAXPROCS higher than 1
==> Starting Consul agent…
==> Starting Consul agent RPC…
==> Consul agent running!
         Node name: 'n2'
        Datacenter: 'dc1'
            Server: true (bootstrap: false)
       Client Addr: 127.0.0.1 (HTTP: 8500, HTTPS: -1, DNS: 8600, RPC: 8400)
      Cluster Addr: 10.10.126.101 (LAN: 8301, WAN: 8302)
    Gossip encrypt: false, RPC-TLS: false, TLS-Incoming: false
             Atlas: <disabled>

==> Log data will now stream in as it occurs:

    2015/07/03 11:30:32 [INFO] serf: EventMemberJoin: n2 10.10.126.101
    2015/07/03 11:30:32 [INFO] serf: EventMemberJoin: n2.dc1 10.10.126.101
    2015/07/03 11:30:32 [INFO] raft: Node at 10.10.126.101:8300 [Follower] entering Follower state
    2015/07/03 11:30:32 [INFO] consul: adding server n2 (Addr: 10.10.126.101:8300) (DC: dc1)
    2015/07/03 11:30:32 [INFO] consul: adding server n2.dc1 (Addr: 10.10.126.101:8300) (DC: dc1)
    2015/07/03 11:30:32 [ERR] agent: failed to sync remote state: No cluster leader
    2015/07/03 11:30:33 [WARN] raft: EnableSingleNode disabled, and no known peers. Aborting election.

从两个server agent的启动日志可以看出,n1、n2启动后并不知道集群其他节点的存在。以n1为例,通过consul members和consul info查看当前agent状态:

$ consul members
Node  Address            Status  Type    Build  Protocol  DC
n1    10.10.105.71:8301  alive   server  0.5.2  2         dc1

$ consul info
… …
consul:
    bootstrap = false
    known_datacenters = 1
    leader = false
    server = true
raft:
    applied_index = 0
    commit_index = 0
    fsm_pending = 0
    last_contact = never
    last_log_index = 0
    last_log_term = 0
    last_snapshot_index = 0
    last_snapshot_term = 0
    num_peers = 0
    state = Follower
    term = 0
… …

可以看出,n1上的agent当前状态是Follower,bootstrap = false;n2同样也是这个情况。整个Cluster并未完成Bootstrap过程。

我们用consul join命令触发Cluster bootstrap过程,我们在n1上执行如下命令:

$ consul join 10.10.126.101
Successfully joined cluster by contacting 1 nodes.

我们通过consul join子命令将当前节点加入包含成员10.10.126.101(也就是n2)的集群中去。命令执行结果通过n1和n2的日志可以观察到:

n1主机:

2015/07/03 09:29:48 [INFO] agent: (LAN) joining: [10.10.126.101]
    2015/07/03 09:29:48 [INFO] serf: EventMemberJoin: n2 10.10.126.101
    2015/07/03 09:29:48 [INFO] agent: (LAN) joined: 1 Err: <nil>
    2015/07/03 09:29:48 [INFO] consul: adding server n2 (Addr: 10.10.126.101:8300) (DC: dc1)
    2015/07/03 09:29:48 [INFO] consul: Attempting bootstrap with nodes: [10.10.126.101:8300 10.10.105.71:8300]
    2015/07/03 09:29:49 [INFO] consul: New leader elected: n2
    2015/07/03 09:29:50 [INFO] agent: Synced service 'consul'

n2主机:

2015/07/03 11:40:53 [INFO] serf: EventMemberJoin: n1 10.10.105.71
    2015/07/03 11:40:53 [INFO] consul: adding server n1 (Addr: 10.10.105.71:8300) (DC: dc1)
    2015/07/03 11:40:53 [INFO] consul: Attempting bootstrap with nodes: [10.10.126.101:8300 10.10.105.71:8300]
    2015/07/03 11:40:54 [WARN] raft: Heartbeat timeout reached, starting election
    2015/07/03 11:40:54 [INFO] raft: Node at 10.10.126.101:8300 [Candidate] entering Candidate state
    2015/07/03 11:40:54 [INFO] raft: Election won. Tally: 2
    2015/07/03 11:40:54 [INFO] raft: Node at 10.10.126.101:8300 [Leader] entering Leader state
    2015/07/03 11:40:54 [INFO] consul: cluster leadership acquired
    2015/07/03 11:40:54 [INFO] consul: New leader elected: n2
    2015/07/03 11:40:54 [INFO] raft: pipelining replication to peer 10.10.105.71:8300
    2015/07/03 11:40:54 [INFO] consul: member 'n2' joined, marking health alive
    2015/07/03 11:40:54 [INFO] consul: member 'n1' joined, marking health alive
    2015/07/03 11:40:55 [INFO] agent: Synced service 'consul'

join后,两台主机互相知道了对方,并进行了leader election过程,n2被选举为Leader。

在n2主机上通过consul info确认一下n2 agent的状态:

$consul info
… …
consul:
    bootstrap = false
    known_datacenters = 1
    leader = true
    server = true
raft:
    applied_index = 10
    commit_index = 10
    fsm_pending = 0
    last_contact = never
    last_log_index = 10
    last_log_term = 1
    last_snapshot_index = 0
    last_snapshot_term = 0
    num_peers = 1
    state = Leader
    term = 1
… …

$ consul members
Node  Address             Status  Type    Build  Protocol  DC
n2    10.10.126.101:8301  alive   server  0.5.2  2         dc1
n1    10.10.105.71:8301   alive   server  0.5.2  2         dc1

可以看到n2的state已经为Leader了,n1的state依旧是Follower。

到这里,n1和n2就成为了dc1这个数据中心Consul Cluster的两个节点,而且是用来维护集群状态的Server node。n2被选举为Leader,n1是Folllower。

如果作为Leader的n2退出集群,我们来看看集群状态会发生怎样变化。在n2上,我们通过consul leave命令告诉n2上的agent离开集群并退出:

$ consul leave
Graceful leave complete

n2上Agent的日志:

2015/07/03 14:04:40 [INFO] agent.rpc: Accepted client: 127.0.0.1:35853
    2015/07/03 14:04:40 [INFO] agent.rpc: Graceful leave triggered
    2015/07/03 14:04:40 [INFO] consul: server starting leave
    2015/07/03 14:04:40 [INFO] raft: Removed peer 10.10.105.71:8300, stopping replication (Index: 7)
    2015/07/03 14:04:40 [INFO] raft: Removed ourself, transitioning to follower
    2015/07/03 14:04:40 [INFO] raft: Node at 10.10.126.101:8300 [Follower] entering Follower state
    2015/07/03 14:04:40 [INFO] serf: EventMemberLeave: n2.dc1 10.10.126.101
    2015/07/03 14:04:40 [INFO] consul: cluster leadership lost
    2015/07/03 14:04:40 [INFO] raft: aborting pipeline replication to peer 10.10.105.71:8300
    2015/07/03 14:04:40 [INFO] consul: removing server n2.dc1 (Addr: 10.10.126.101:8300) (DC: dc1)
    2015/07/03 14:04:41 [INFO] serf: EventMemberLeave: n2 10.10.126.101
    2015/07/03 14:04:41 [INFO] consul: removing server n2 (Addr: 10.10.126.101:8300) (DC: dc1)
    2015/07/03 14:04:41 [INFO] agent: requesting shutdown
    2015/07/03 14:04:41 [INFO] consul: shutting down server
    2015/07/03 14:04:42 [INFO] agent: shutdown complete

n1上的日志:

2015/07/03 11:53:36 [INFO] serf: EventMemberLeave: n2 10.10.126.101
2015/07/03 11:53:36 [INFO] consul: removing server n2 (Addr: 10.10.126.101:8300) (DC: dc1)
2015/07/03 11:55:15 [ERR] agent: failed to sync remote state: No cluster leader

这个时候我们在n1上通过consul info查看,n1的状态依旧是Follower,也就是说在双server节点的集群下,一个server退出,将产生无Leader状态。在三 server节点集群里,Leader退出,其余两个会再协商选出一个新Leader,但一旦再退出一个节点,同样集群就不会再有Leader了。 当然,如果是单节点bootstrap的集群( -bootstrap-expect 1 ),集群只有一个server节点,那这个server节点自然当选Leader。

现在我们在n1上通过consul members查看集群状态:

$ consul members
Node  Address             Status  Type    Build  Protocol  DC
n1    10.10.105.71:8301   alive   server  0.5.2  2         dc1
n2    10.10.126.101:8301  left    server  0.5.2  2         dc1

执行结果显示:n2是Left状态。我们重新启动n2,再来看看集群的状态变化。

$ consul agent -server -bootstrap-expect 2 -data-dir /tmp/consul -node=n2 -bind=10.10.126.101 -ui-dir ./dist  -dc=dc1
… …
==> Log data will now stream in as it occurs:

    2015/07/03 14:13:46 [INFO] serf: EventMemberJoin: n2 10.10.126.101
    2015/07/03 14:13:46 [INFO] raft: Node at 10.10.126.101:8300 [Follower] entering Follower state
    2015/07/03 14:13:46 [INFO] consul: adding server n2 (Addr: 10.10.126.101:8300) (DC: dc1)
    2015/07/03 14:13:46 [INFO] serf: EventMemberJoin: n2.dc1 10.10.126.101
    2015/07/03 14:13:46 [INFO] consul: adding server n2.dc1 (Addr: 10.10.126.101:8300) (DC: dc1)
    2015/07/03 14:13:46 [ERR] agent: failed to sync remote state: No cluster leader
    2015/07/03 14:13:48 [WARN] raft: EnableSingleNode disabled, and no known peers. Aborting election.
… …

n2启动后,并未自动加入之前的cluster,而是依旧如第一次启动那样,看不到peers,孤立运行。

我们再来在n1上join一下:consul join 10.10.126.101

n1的日志变为:

2015/07/03 12:04:55 [INFO] consul: adding server n2 (Addr: 10.10.126.101:8300) (DC: dc1)
2015/07/03 12:04:56 [ERR] agent: failed to sync remote state: No cluster leader

n2的日志变为:

    2015/07/03 14:16:00 [INFO] serf: EventMemberJoin: n1 10.10.105.71
    2015/07/03 14:16:00 [INFO] consul: adding server n1 (Addr: 10.10.105.71:8300) (DC: dc1)
    2015/07/03 14:16:00 [INFO] consul: New leader elected: n2
    2015/07/03 14:16:01 [ERR] agent: failed to sync remote state: No cluster leader

n1和n2无法再选出Leader,通过info命令看,两个节点都变成了Follower,集群仍然处于无Leader状态。

这个问题在consul的github repositroy issues中被多人多次提及,但作者似乎不将此作为bug。产生这个问题的原因是当n2退出时,consul会将/tmp/consul/raft /peers.json的内容由:

["10.10.105.71:8300", "10.10.126.101:8300"]

改为

null

n2重启后,该文件并未改变,依旧为null,n2启动就不会重新自动join到n1的cluster中。

关于这个问题的cluster恢复方法,官方在Outage Recovery一文中有明确说明。我们来测试一下:

我们打开n1和n2的/tmp/consul/raft/peers.json,将其内容统一修改为:

["10.10.126.101:8300","10.10.105.71:8300"]

然后重启n2,但加上-rejoin命令:

$ consul agent -server -bootstrap-expect 2 -data-dir /tmp/consul -node=n2 -bind=10.10.126.101 -ui-dir ./dist  -dc=dc1 -rejoin

…. …

    2015/07/03 14:56:02 [WARN] raft: Election timeout reached, restarting election
    2015/07/03 14:56:02 [INFO] raft: Node at 10.10.126.101:8300 [Candidate] entering Candidate state
    2015/07/03 14:56:02 [INFO] raft: Election won. Tally: 2
    2015/07/03 14:56:02 [INFO] raft: Node at 10.10.126.101:8300 [Leader] entering Leader state
    2015/07/03 14:56:02 [INFO] consul: cluster leadership acquired
    2015/07/03 14:56:02 [INFO] consul: New leader elected: n2

…….

n1上的日志:

2015/07/03 12:44:52 [INFO] serf: EventMemberJoin: n2 10.10.126.101
    2015/07/03 12:44:52 [INFO] consul: adding server n2 (Addr: 10.10.126.101:8300) (DC: dc1)
    2015/07/03 12:44:54 [INFO] consul: New leader elected: n2
    2015/07/03 12:44:55 [WARN] raft: Rejecting vote from 10.10.126.101:8300 since we have a leader: 10.10.126.101:8300
    2015/07/03 12:44:56 [WARN] raft: Heartbeat timeout reached, starting election
    2015/07/03 12:44:56 [INFO] raft: Node at 10.10.105.71:8300 [Candidate] entering Candidate state
    2015/07/03 12:44:56 [ERR] raft: Failed to make RequestVote RPC to 10.10.126.101:8300: EOF
    2015/07/03 12:44:57 [INFO] raft: Node at 10.10.105.71:8300 [Follower] entering Follower state
    2015/07/03 12:44:57 [INFO] consul: New leader elected: n2

这回集群的Leader重新选举成功,集群状态恢复。

接下来我们启动n3上的client mode agent:

$ consul agent  -data-dir /tmp/consul -node=n3 -bind=10.10.126.187  -dc=dc1
==> WARNING: It is highly recommended to set GOMAXPROCS higher than 1
==> Starting Consul agent…
==> Starting Consul agent RPC…
==> Consul agent running!
         Node name: 'n3'
        Datacenter: 'dc1'
            Server: false (bootstrap: false)
       Client Addr: 127.0.0.1 (HTTP: 8500, HTTPS: -1, DNS: 8600, RPC: 8400)
      Cluster Addr: 10.10.126.187 (LAN: 8301, WAN: 8302)
    Gossip encrypt: false, RPC-TLS: false, TLS-Incoming: false
             Atlas: <disabled>

==> Log data will now stream in as it occurs:

    2015/07/03 14:55:17 [INFO] serf: EventMemberJoin: n3 10.10.126.187
    2015/07/03 14:55:17 [ERR] agent: failed to sync remote state: No known Consul servers

在n3上join n1后,n3的日志输出如下:

   2015/07/03 14:59:31 [INFO] agent: (LAN) joining: [10.10.105.71]
    2015/07/03 14:59:31 [INFO] serf: EventMemberJoin: n2 10.10.126.101
    2015/07/03 14:59:31 [INFO] serf: EventMemberJoin: n1 10.10.105.71
    2015/07/03 14:59:31 [INFO] agent: (LAN) joined: 1 Err: <nil>
    2015/07/03 14:59:31 [INFO] consul: adding server n2 (Addr: 10.10.126.101:8300) (DC: dc1)
    2015/07/03 14:59:31 [INFO] consul: adding server n1 (Addr: 10.10.105.71:8300) (DC: dc1)

n3上consul members可以查看到如下内容:

$ consul members
Node  Address             Status  Type    Build  Protocol  DC
n1    10.10.105.71:8301   alive   server  0.5.2  2         dc1
n3    10.10.126.187:8301  alive   client  0.5.2  2         dc1
n2    10.10.126.101:8301  alive   server  0.5.2  2         dc1

处于client mode的agent可以自由退出和启动,不会出现server mode下agent的问题。

二、服务注册与发现

我们建立Consul Cluster是为了实现服务的注册和发现。Consul支持两种服务注册的方式,一种是通过Consul的服务注册HTTP API,由服务自身在启动后调用API注册自己,另外一种则是通过在配置文件中定义服务的方式进行注册。Consul文档中建议使用后面一种方式来做服务 配置和服务注册。

我们还是用例子来说明一下如何做服务配置。前面我们已经建立了Consul Cluster,Cluster里包含了三个Node:两个Server mode node,一个Client mode Node。我们计划在n2、n3上部署一类服务web3,于是我们需要分别在n2、n3上增加Consul agent的配置文件。

Consul agent在启动时可以通过-config-dir来指定配置文件所在目录,比如以n3为例,我们可以如此启动n3:

consul agent -data-dir /tmp/consul -node=n3 -bind=10.10.126.187 -dc=dc1 -config-dir=./conf

这样在./conf下的所有文件扩展为.json的文件都会被Consul agent作为配置文件读取。

我们以n3为例,我们在n3的consul agent的配置文件目录下创建web3.json文件:

//web3.json
{
  "service": {
    "name": "web3",
    "tags": ["master"],
    "address": "127.0.0.1",
    "port": 10000,
    "checks": [
      {
        "http": "http://localhost:10000/health",
        "interval": "10s"
      }
    ]
  }
}

这个配置就是我们在n3节点上为web3这个服务做的服务定义,定义中包含服务的name、address、port等,还包含一个服务检测的配置,这里 我们每隔10s对服务进行一次健康检查,这要求服务增加对/health的处理逻辑。同理,我们在n2上也建立同样配置文件(n2需重启,并带上 -config-dir命令行选项),服务注册就这么简单。

在重启后的n2、n3日志中,我们能发现如下的错误内容:

2015/07/06 13:48:11 [WARN] agent: http request failed 'http://localhost:10000/health' : Get http://localhost:10000/health: dial tcp 127.0.0.1:10000: connect failed"

这就是agent对定义的服务的check日志。为了避免这个错误日志刷屏,我们在n2、n3上各部署一个web3服务实例。以n3上的web3为例,其源码如下:

//web3.go
package main

import (
    "fmt"
    "net/http"
)

func handler(w http.ResponseWriter, r *http.Request) {
    fmt.Println("hello Web3! This is n3")
    fmt.Fprintf(w, "Hello Web3! This is n3")
}

func healthHandler(w http.ResponseWriter, r *http.Request) {
    fmt.Println("health check!")
}

func main() {
    http.HandleFunc("/", handler)
    http.HandleFunc("/health", healthHandler)
    http.ListenAndServe(":10000", nil)
}

一旦n2、n3上的web3服务实例启动,我们就可以尝试发现这些服务了。

Consul提供了两种发现服务的方式,一种是通过HTTP API查看存在哪些服务;另外一种是通过consul agent内置的DNS服务来做。两者的差别在于后者可以根据服务check的实时状态动态调整available服务节点列表。我们这里也着重说明适用 DNS方式进行服务发现的具体步骤。

在配置和部署完web3服务后,我们就可以通过DNS命令来查询服务的具体信息了。consul为服务编排的内置域名为 “NAME.service.consul",这样我们的web3的域名为:web3.service.consul。我们在n1通过dig工具来查看一 下,注意是在n1上,n1上并未定义和部署web3服务,但集群中服务的信息已经被同步到n1上了,信息是一致的:

$ dig @127.0.0.1 -p 8600 web3.service.consul SRV

; <<>> DiG 9.9.5-3-Ubuntu <<>> @127.0.0.1 -p 8600 web3.service.consul SRV
; (1 server found)
;; global options: +cmd
;; Got answer:
;; ->>HEADER<<- opcode: QUERY, status: NOERROR, id: 6713
;; flags: qr aa rd; QUERY: 1, ANSWER: 2, AUTHORITY: 0, ADDITIONAL: 2
;; WARNING: recursion requested but not available

;; QUESTION SECTION:
;web3.service.consul.        IN    SRV

;; ANSWER SECTION:
web3.service.consul.    0    IN    SRV    1 1 10000 n2.node.dc1.consul.
web3.service.consul.    0    IN    SRV    1 1 10000 n3.node.dc1.consul.

;; ADDITIONAL SECTION:
n2.node.dc1.consul.    0    IN    A    127.0.0.1
n3.node.dc1.consul.    0    IN    A    127.0.0.1

;; Query time: 2 msec
;; SERVER: 127.0.0.1#8600(127.0.0.1)
;; WHEN: Mon Jul 06 12:12:53 CST 2015
;; MSG SIZE  rcvd: 219

可以看到在ANSWER SECTION中,我们得到了两个结果:n2和n3上各有一个web3的服务。在dig命令中我们用了SRV标志,那是因为我们需要的服务信息不仅有ip地址,还需要有端口号。

现在我们停掉n2上的web3服务,10s后,我们再来查一下:

$ dig @127.0.0.1 -p 8600 web3.service.consul SRV

; <<>> DiG 9.9.5-3-Ubuntu <<>> @127.0.0.1 -p 8600 web3.service.consul SRV
; (1 server found)
;; global options: +cmd
;; Got answer:
;; ->>HEADER<<- opcode: QUERY, status: NOERROR, id: 25136
;; flags: qr aa rd; QUERY: 1, ANSWER: 1, AUTHORITY: 0, ADDITIONAL: 1
;; WARNING: recursion requested but not available

;; QUESTION SECTION:
;web3.service.consul.        IN    SRV

;; ANSWER SECTION:
web3.service.consul.    0    IN    SRV    1 1 10000 n3.node.dc1.consul.

;; ADDITIONAL SECTION:
n3.node.dc1.consul.    0    IN    A    127.0.0.1

;; Query time: 3 msec
;; SERVER: 127.0.0.1#8600(127.0.0.1)
;; WHEN: Mon Jul 06 12:16:39 CST 2015
;; MSG SIZE  rcvd: 128

结果显示,只有n3上这一个web3服务可用了。通过下面Consul Agent日志:

dns: node 'n2' failing health check 'service web3' check', dropping from service 'web3'

我们可以看到consul agent将health check失败的web3从结果列表中剔除了,这样web3服务的客户端在服务发现过程中就只能获取到当前可用的web3服务节点了,这个好处是在实际应 用中大大降低了客户端实现”服务发现“时的难度。另外consul agent DNS在返回查询结果时也支持DNS Server常见的策略,至少是支持轮询。你可以多次执行dig命令,可以看到n2和n3的排列顺序是不同的。还有一点值得注意的是:由于考虑DNS cache对consul agent查询结果的影响,默认情况下所有由consul agent返回的结果TTL值均设为0,也就是说不支持dns结果缓存。

接下来,我们使用golang实现一个demo级别的服务发现的客户端,这里会用到第三方dns client库"github.com/miekg/dns"。

// servicediscovery.go
package main

import (
    "fmt"
    "log"

    "github.com/miekg/dns"
)

const (
        srvName = "web3.service.consul"
        agentAddr = "127.0.0.1:8600"
)

func main() {
    c := new(dns.Client)

    m := new(dns.Msg)
    m.SetQuestion(dns.Fqdn(srvName), dns.TypeSRV)
    m.RecursionDesired = true

    r, _, err := c.Exchange(m, agentAddr)
    if r == nil {
        log.Fatalf("dns query error: %s\n", err.Error())
    }

    if r.Rcode != dns.RcodeSuccess {
        log.Fatalf("dns query error: %v\n", r.Rcode)
    }
   
    for _, a := range r.Answer {
        b, ok := a.(*dns.SRV)
        if ok {
            m.SetQuestion(dns.Fqdn(b.Target), dns.TypeA)
            r1, _, err := c.Exchange(m, agentAddr)
            if r1 == nil {
                log.Fatalf("dns query error: %v, %v\n", r1.Rcode, err)
            }
            for _, a1 := range r1.Answer {
                c, ok := a1.(*dns.A)
                if ok {
                   fmt.Printf("%s – %s:%d\n", b.Target, c.A, b.Port)
                }
            }
        }
    }
}

我们执行该程序:
$ go run servicediscovery.go
n2.node.dc1.consul. – 10.10.126.101:10000
n3.node.dc1.consul. – 10.10.126.187:10000

注意各个node上的服务check是由其node上的agent上进行的,一旦那个node上的agent出现问题,则位于那个node上的所有 service也将会被置为unavailable状态。比如我们停掉n3上的agent,那么我们在进行web3服务节点查询时,就只能获取到n2这一 个节点上有可用的web3服务了。

在真实的程序中,我们可以像上面demo中那样,每Request都做一次DNS查询,不过这样的代价也很高。稍复杂些,我们可以结合dns结果本地缓存+定期查询+每遇到Failed查询的方式来综合考量服务的发现方法或利用Consul提供的watch命令等。

以上仅仅是Consul的一个入门。真实场景中,理想的方案需要考虑的事情还有很多。Consul自身目前演进到0.5.2版本,还有不完善之处,但它已 经被很多公司用于production环境。Consul不是孤立的,要充分发挥出Consul的优势,在真实方案中,我们还要考虑与 Docker,HAProxy,Mesos等工具的结合。

如发现本站页面被黑,比如:挂载广告、挖矿等恶意代码,请朋友们及时联系我。十分感谢! Go语言第一课 Go语言精进之路1 Go语言精进之路2 商务合作请联系bigwhite.cn AT aliyun.com

欢迎使用邮件订阅我的博客

输入邮箱订阅本站,只要有新文章发布,就会第一时间发送邮件通知你哦!

这里是 Tony Bai的个人Blog,欢迎访问、订阅和留言! 订阅Feed请点击上面图片

如果您觉得这里的文章对您有帮助,请扫描上方二维码进行捐赠 ,加油后的Tony Bai将会为您呈现更多精彩的文章,谢谢!

如果您希望通过微信捐赠,请用微信客户端扫描下方赞赏码:

如果您希望通过比特币或以太币捐赠,可以扫描下方二维码:

比特币:

以太币:

如果您喜欢通过微信浏览本站内容,可以扫描下方二维码,订阅本站官方微信订阅号“iamtonybai”;点击二维码,可直达本人官方微博主页^_^:
本站Powered by Digital Ocean VPS。
选择Digital Ocean VPS主机,即可获得10美元现金充值,可 免费使用两个月哟! 著名主机提供商Linode 10$优惠码:linode10,在 这里注册即可免费获 得。阿里云推荐码: 1WFZ0V立享9折!


View Tony Bai's profile on LinkedIn
DigitalOcean Referral Badge

文章

评论

  • 正在加载...

分类

标签

归档



View My Stats