• 如果您觉得本站非常有看点,那么赶紧使用Ctrl+D 收藏吧

分布式系统选主场景分析及实现

开发技术 开发技术 3周前 (09-09) 24次浏览

一:需要选主的场景
1:服务有多台机器,取其中一台去执行任务。多台机器同时执行会出问题,如将数据库中状态为失败的记录取出来重新执行,如果多台机器同时执行,会导致一个失败的任务被多台机器同时执行。
2:服务有多台机器,选其中一台作为主,主负责任务的分发,大家一起消费并处理任务。还是将数据库中状态为失败的记录取出来重新执行,由于一台机器可能处理不过来,需要多台机器协同处理。这个时候主机器负责将失败的记录从数据库中查出来,写入消息队列,其他机器一同消费队列中的任务,并处理失败的记录
 
二:进行选主
根据上面的选主场景,我们其实可以从多台机器中随机取一台,比raft这种选主算法简单得多。我们甚至可以在配置文件中指定一台机器,只有这台机器才执行相关功能,其他机器则不执行。如果是固定的几台机器,且一台机器也能完成我们的需求,这样搞其实也可以。如果机器不固定,而且单台处理不过来时,用配置文件的方式就不适合。
可采用竞争选主的方式,谁先抢到谁就是主。
 
1:方案一
采用redis方案实现。如果指定的key不存在就将机器信息写入这个key,成功写入的那台机器就是主,设置过期时间,防止机器异常挂掉的情况,所有的机器都需要定时去抢redis锁。SETNX这个命令就满足我们的需求,写redis成功的就是主,写失败的就是从。
优点:
  • 1:实现简单,比配置文件的方式好一点,支持机器动态
缺点:
  • 1:需要定时去抢锁
  • 2:主可能经常变化,而且要保证主在切换的过程中业务逻辑的正确性
  • 3:有些时间片可能没有主,就是主挂掉了,而其他机器还没到抢锁的时间,这个时间片就没有主
 
2:方案二
采用etcd方案实现。etcd支持事务能做到不存在就写入,达到redis SETNX一样的效果,而且通过etcd的租赁机制保证在主挂掉的情况下通知所有机器,这时大家自动开始新一轮的选主,还是那句话第一个抢到的就是主。
优点:
  • 满足我们的需求,没有设计上的缺陷
  • 只有主挂掉的情况,才会重新选主,不用担心主在切换的过程中对业务逻辑的影响
缺点:
  • 实现起来相对复杂,那我就来试试吧 

golang源码实现如下:

  1 package etcdDemo
  2 
  3 import (
  4     "context"
  5     "fmt"
  6     "github.com/coreos/etcd/clientv3"
  7     "github.com/google/uuid"
  8     "time"
  9 )
 10 
 11 type Callback func(isMaster bool)
 12 
 13 type SelectMaster struct {
 14     endPoints []string
 15     key       string
 16     cli       *clientv3.Client
 17     lease     *clientv3.LeaseGrantResponse
 18     chClose   chan int
 19     callback  Callback
 20     token     string
 21     isMaster  bool
 22 }
 23 
 24 func NewSelectMaster(endPoints []string, key string) (*SelectMaster, error) {
 25     sm := &SelectMaster{
 26         endPoints: endPoints,
 27         key:       key,
 28         chClose:   make(chan int, 0),
 29         token:     uuid.New().String(),
 30     }
 31 
 32     cli, err := clientv3.New(clientv3.Config{
 33         Endpoints:   endPoints,
 34         DialTimeout: 3 * time.Second,
 35     })
 36     if err != nil {
 37         return sm, err
 38     }
 39     sm.cli = cli
 40     go sm.ioLoop()
 41     return sm, nil
 42 }
 43 
 44 func (sm *SelectMaster) ioLoop() {
 45     fmt.Println("SelectMaster.ioLoop start")
 46     ticker := time.NewTicker(time.Second * 3)
 47     defer ticker.Stop()
 48     chWatch := sm.cli.Watch(context.TODO(), sm.key)
 49     for {
 50         select {
 51         case <-ticker.C:
 52             if sm.lease == nil {
 53                 leaseResp, err := sm.cli.Grant(context.Background(), 4)
 54                 if err != nil {
 55                     fmt.Println("cli.Grant error=", err.Error())
 56                 } else {
 57                     sm.lease = leaseResp
 58                 }
 59             }
 60             if sm.lease != nil {
 61                 _, err := sm.cli.KeepAliveOnce(context.Background(), sm.lease.ID)
 62                 if err != nil {
 63                     fmt.Println("cli.KeepAliveOnce error=", err.Error())
 64                     break
 65                 }
 66             }
 67         case c := <-chWatch:
 68             for _, e := range c.Events {
 69                 if e == nil || e.Kv == nil {
 70                     continue
 71                 }
 72                 token := string(e.Kv.Value)
 73                 sm.isMaster = sm.token == token
 74                 if sm.callback == nil {
 75                     fmt.Println("SelectMaster.callback is nil")
 76                 } else {
 77                     sm.callback(sm.isMaster)
 78                     fmt.Println("SelectMaster.isLoop token=", token)
 79                     if token == "" { //主挂了,开始竞选
 80                         sm.election()
 81                     }
 82                 }
 83             }
 84         case <-sm.chClose:
 85             goto stop
 86         }
 87     }
 88 stop:
 89     fmt.Println("SelectMaster.ioLoop end")
 90 }
 91 
 92 func (sm *SelectMaster) IsMaster() bool {
 93     return sm.isMaster
 94 }
 95 
 96 func (sm *SelectMaster) Close() {
 97     sm.chClose <- 1
 98 }
 99 
100 func (sm *SelectMaster) Election(callback Callback) (bool, error) {
101     sm.callback = callback
102     return sm.election()
103 }
104 
105 func (sm *SelectMaster) election() (bool, error) {
106     ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
107     defer cancel()
108     leaseResp, err := sm.cli.Grant(ctx, 10)
109     if err != nil {
110         return false, err
111     }
112     sm.lease = leaseResp
113     txn := clientv3.NewKV(sm.cli).Txn(context.TODO())
114     txn.If(clientv3.Compare(clientv3.CreateRevision(sm.key), "=", 0)).
115         Then(clientv3.OpPut(sm.key, sm.token, clientv3.WithLease(leaseResp.ID))).Else()
116     txnResp, err := txn.Commit()
117     if err != nil {
118         return false, err
119     }
120     return txnResp.Succeeded, nil
121 }
122 
123 func testSelectMaster() *SelectMaster {
124     endPoints := []string{"172.25.20.248:2379"}
125     sm, err := NewSelectMaster(endPoints, "/test/lock")
126     if err != nil {
127         fmt.Println(err.Error())
128         return nil
129     }
130     callback := func(isMaster bool) {
131         fmt.Println(sm.token, "callback=", isMaster)
132     }
133     isSuccess, err := sm.Election(callback)
134     if err != nil {
135         fmt.Println(sm.token, "Election=", err.Error())
136     } else {
137         fmt.Println(sm.token, "Election=", isSuccess)
138     }
139     return sm
140 }
141 
142 func TestSelectMaster() {
143     var master *SelectMaster
144     for i := 0; i < 3; i++ {
145         sm := testSelectMaster()
146         if sm.IsMaster() {
147             master = sm
148         }
149     }
150     if master != nil {
151         master.Close()
152     }
153     time.Sleep(time.Second*10)
154 }

 


喜欢 (0)