Etcd使用指南：用分布式锁思路轻松实现自动选主

Etcd是一款开源分布式键值存储系统，具备高可用、强一致性和高性能等特点，非常适合作为分布式锁的存储后端。

分布式锁可以保证当有多台实例同时竞争一把锁时，只有一个人会成功，其他的都是失败。诸如共享资源修改、幂等、频控等场景都可以通过分布式锁来实现。

还有一种场景，也可以通过分布式锁来实现，就是自动选主。

自动选主是指在分布式系统中，通过某种机制来选出一个主节点，这个主节点可以承担一些特殊的任务，比如数据写入、资源分配等。

Etcd实现自动选主的思路

Etcd实现自动选主的思路很简单，就是利用分布式锁的特性来实现。

具体步骤如下：

1. 在Etcd中创建一个名为“lock”的键，并将其值设置为一个随机生成的字符串。
2. 各个实例同时尝试获取“lock”键的锁，只有成功获取锁的实例才能成为主节点。
3. 主节点在一定时间内保持锁的状态，并在锁过期后释放锁。
4. 其他实例不断尝试获取锁，直到成功获取锁成为新的主节点。

Etcd实现自动选主的代码示例

import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import io.etcd.jetcd.ByteSequence;
import io.etcd.jetcd.Client;
import io.etcd.jetcd.KV;
import io.etcd.jetcd.Lease;
import io.etcd.jetcd.lease.LeaseGrantResponse;
import io.etcd.jetcd.lock.LockResponse;
import io.etcd.jetcd.options.GetOption;
import io.etcd.jetcd.options.LeaseOption;
import io.etcd.jetcd.options.PutOption;
import io.etcd.jetcd.watch.WatchEvent;
import io.etcd.jetcd.watch.WatchResponse;
import java.nio.charset.Charset;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

public class EtcdAutoLeader {

  private static final ByteSequence LOCK_KEY = ByteSequence.from("lock", Charset.defaultCharset());

  public static void main(String[] args) throws Exception {
    // 创建Etcd客户端
    Client client = Client.create();

    // 创建一个线程池，用于定时续约锁
    ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("lease-renew-thread-%d").build();
    ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1, threadFactory);

    // 创建一个KV客户端，用于操作键值数据
    KV kvClient = client.getKVClient();

    // 创建一个租约客户端，用于管理租约
    Lease leaseClient = client.getLeaseClient();

    // 尝试获取锁
    LockResponse lockResponse = kvClient.lock(LOCK_KEY).execute();

    // 如果获取锁成功，则成为主节点
    if (lockResponse.succeeded()) {
      System.out.println("获取锁成功，成为主节点");

      // 创建一个租约，用于续约锁
      LeaseGrantResponse leaseGrantResponse = leaseClient.grant(30, TimeUnit.SECONDS, LeaseOption.newBuilder().build()).get();

      // 定时续约锁
      executorService.scheduleAtFixedRate(() -> {
        try {
          leaseClient.renew(leaseGrantResponse.getID()).get();
        } catch (Exception e) {
          e.printStackTrace();
        }
      }, 0, 10, TimeUnit.SECONDS);

      // 监听锁的变化，如果锁被释放，则重新尝试获取锁
      kvClient.watch(LOCK_KEY, GetOption.newBuilder().withPrefix(true).build()).get().listener(MoreExecutors.directExecutor()).listen(watchResponse -> {
        if (watchResponse.getEventsList().stream().anyMatch(event -> event.getEventType() == WatchEvent.EventType.DELETE)) {
          System.out.println("锁被释放，重新尝试获取锁");
          try {
            LockResponse newLockResponse = kvClient.lock(LOCK_KEY).execute();
            if (newLockResponse.succeeded()) {
              System.out.println("获取锁成功，成为新的主节点");
            }
          } catch (Exception e) {
            e.printStackTrace();
          }
        }
      });

      // 运行主节点任务
      while (true) {
        // 主节点任务代码
      }
    } else {
      System.out.println("获取锁失败，成为从节点");

      // 监听锁的变化，如果锁被释放，则重新尝试获取锁
      kvClient.watch(LOCK_KEY, GetOption.newBuilder().withPrefix(true).build()).get().listener(MoreExecutors.directExecutor()).listen(watchResponse -> {
        if (watchResponse.getEventsList().stream().anyMatch(event -> event.getEventType() == WatchEvent.EventType.DELETE)) {
          System.out.println("锁被释放，重新尝试获取锁");
          try {
            LockResponse newLockResponse = kvClient.lock(LOCK_KEY).execute();
            if (newLockResponse.succeeded()) {
              System.out.println("获取锁成功，成为新的主节点");
            }
          } catch (Exception e) {
            e.printStackTrace();