使用etcd实现动态分布式选主

你的名字 2023-03-02 05:04 16阅读 0赞

日常开发中经常会有后台运行的worker类任务，由于服务是分布式的，我们可能会有多个分布式的worker同时在运行，有时候我们需要分布式下只有一个worker在运行，这时候就可以用到etcd的分布式选主。

![20200727194126640.png][]

etcd中concurrency包下已经帮我们实现好了选主，我们只需要调用其api实现就可以了，下面我们分析下etcd是如何实现选主机制的。直接进行源码分析：

// Campaign puts a value as eligible for the election on the prefix
    // key.
    // Multiple sessions can participate in the election for the
    // same prefix, but only one can be the leader at a time.
    //
    // If the context is 'context.TODO()/context.Background()', the Campaign
    // will continue to be blocked for other keys to be deleted, unless server
    // returns a non-recoverable error (e.g. ErrCompacted).
    // Otherwise, until the context is not cancelled or timed-out, Campaign will
    // continue to be blocked until it becomes the leader.
    
    // 多个etcd的session可以通过prefix来参与选举。但是只有一个session能成为leader。
    // Campaign方法会阻塞，直到session成功成为leader才返回。
    func (e *Election) Campaign(ctx context.Context, val string) error {
    	s := e.session
    	client := e.session.Client()
    	// 根据前缀和租约创建当前key
    	k := fmt.Sprintf("%s%x", e.keyPrefix, s.Lease())
    	// 如果是第一次创建key，那么key的revision为0
    	// 这里用到了etcd的事务，如果if判断为true，那么put这个key，否则get这个key；最终都能获取到这个key的内容。
    	txn := client.Txn(ctx).If(v3.Compare(v3.CreateRevision(k), "=", 0))
    	txn = txn.Then(v3.OpPut(k, val, v3.WithLease(s.Lease())))
    	txn = txn.Else(v3.OpGet(k))
    	resp, err := txn.Commit()
    	if err != nil {
    		return err
    	}
    	e.leaderKey, e.leaderRev, e.leaderSession = k, resp.Header.Revision, s
    	// 这里是事务中if判断为false，即执行了else
    	if !resp.Succeeded {
    		kv := resp.Responses[0].GetResponseRange().Kvs[0]
    		e.leaderRev = kv.CreateRevision
    		if string(kv.Value) != val { // 判定val是否相同，不相同的话，在不更换leader的情况下，更新val
    			if err = e.Proclaim(ctx, val); err != nil {
    				e.Resign(ctx)
    				return err
    			}
    		}
    	}
    	// 等待prefix前缀下所有比当前key的revision小的其他key都被删除后，才返回，竞选为leader
    	_, err = waitDeletes(ctx, client, e.keyPrefix, e.leaderRev-1)
    	if err != nil {
    		// clean up in case of context cancel
    		select {
    		case <-ctx.Done():
    			e.Resign(client.Ctx())
    		default:
    			e.leaderSession = nil
    		}
    		return err
    	}
    	e.hdr = resp.Header
    
    	return nil
    }

func waitDelete(ctx context.Context, client *v3.Client, key string, rev int64) error {
    	cctx, cancel := context.WithCancel(ctx)
    	defer cancel()
    
    	var wr v3.WatchResponse
    	// 这里watch指定的key，对于这个key所有events事件，都会收到服务端的推送。
    	wch := client.Watch(cctx, key, v3.WithRev(rev))
    	for wr = range wch {
    		for _, ev := range wr.Events {
    			if ev.Type == mvccpb.DELETE { // 如果当前这个key被删除了，那么会退出这个方法，watch下一个key。
    				return nil
    			}
    		}
    	}
    	if err := wr.Err(); err != nil {
    		return err
    	}
    	if err := ctx.Err(); err != nil {
    		return err
    	}
    	return fmt.Errorf("lost watcher waiting for delete")
    }
    
    // waitDeletes efficiently waits until all keys matching the prefix and no greater
    // than the create revision.
    func waitDeletes(ctx context.Context, client *v3.Client, pfx string, maxCreateRev int64) (*pb.ResponseHeader, error) {
            // option，获取createRevision不大于maxCreateRev的key，只取最后一个revision最大的
    	getOpts := append(v3.WithLastCreate(), v3.WithMaxCreateRev(maxCreateRev))
    	for {
    		// 获取前缀prefix下，所有比指定revision小的key
    		resp, err := client.Get(ctx, pfx, getOpts...)
    		if err != nil {
    			return nil, err
    		}
    		if len(resp.Kvs) == 0 {
    			return resp.Header, nil
    		}
    		lastKey := string(resp.Kvs[0].Key)
    		// 去watch revision最大的key，这里也会阻塞的watch。外层有循环判断，要等所有比revision小的key的没了，才退出。
                    // 下方有具体的说明
    		if err = waitDelete(ctx, client, lastKey, resp.Header.Revision); err != nil {
    			return nil, err
    		}
    	}
    }

总体来说还是比较好理解的，主要是利用watch机制来实现了节点在不是leader的时候的阻塞机制。

可以看到，每个节点都创建了自己的key，但是这些key的前缀是一致的，选主是根据前缀去选主的。

如果有a，b，c三个节点同时去竞选，分别对应竞选的createRevision是0，1，2，那么每个节点会watch比自己createRevision**小并且最大**的节点，这是个循环的过程，等到**所有**比自己createRevision小的节点都被删除后，自己才成为leader。

对应的，a节点会成为leader，b节点在watch a节点，c节点在watch b节点。如果b节点key被删除了，c节点会去watch a节点。

如果a节点key被删除了，b节点会成为leader。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpeXVubG9uZzQx_size_16_color_FFFFFF_t_70][]

TODO：后面补充如何使用……

好了，现在补充如何使用etcd提供的这个包来实现分布式下选主。

直接贴代码吧，然后注释里会讲解。

[20200727194126640.png]: /images/20230208/789b60ffdd39400eb67a1fdcccac0fd9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpeXVubG9uZzQx_size_16_color_FFFFFF_t_70]: /images/20230208/62b30e9e1409427f9f27ca83b865928a.png