java - 如何使用 Curator for Zookeeper 有效地使用 LeaderElection 配方？

Question

我正在使用 Apache Curator 库在 Zookeeper 上进行领导选举。我将我的应用程序代码部署在不同的机器上，我只需要从一台机器上执行我的代码，这就是我在 zookeeper 上进行领导选举的原因，这样我就可以检查我是否是领导者，然后执行这段代码。

下面是我的 LeaderElectionExecutor 类，它确保每个应用程序都有一个 Curator 实例

public class LeaderElectionExecutor {

    private ZookeeperClient zookClient;

    private static final String LEADER_NODE = "/testleader";

    private static class Holder {
        static final LeaderElectionExecutor INSTANCE = new LeaderElectionExecutor();
    }

    public static LeaderElectionExecutor getInstance() {
        return Holder.INSTANCE;
    }

    private LeaderElectionExecutor() {
        try {
            String hostname = Utils.getHostName();

            String nodes = "host1:2181,host2:2181;

            zookClient = new ZookeeperClient(nodes, LEADER_NODE, hostname);
            zookClient.start();

            // added sleep specifically for the leader to get selected
            // since I cannot call isLeader method immediately after starting the latch
            TimeUnit.MINUTES.sleep(1);
        } catch (Exception ex) {
            // logging error
            System.exit(1);
        }
    }

    public ZookeeperClient getZookClient() {
        return zookClient;
    }
}

下面是我的 ZookeeperClient 代码 -

// can this class be improved in any ways?
public class ZookeeperClient {

    private CuratorFramework client;
    private String latchPath;
    private String id;
    private LeaderLatch leaderLatch;

    public ZookeeperClient(String connString, String latchPath, String id) {
        client = CuratorFrameworkFactory.newClient(connString, new ExponentialBackoffRetry(1000, Integer.MAX_VALUE));
        this.id = id;
        this.latchPath = latchPath;
    }

    public void start() throws Exception {
        client.start();
        leaderLatch = new LeaderLatch(client, latchPath, id);
        leaderLatch.start();
    }

    public boolean isLeader() {
        return leaderLatch.hasLeadership();
    }

    public Participant currentLeader() throws Exception {
        return leaderLatch.getLeader();
    }

    public void close() throws IOException {
        leaderLatch.close();
        client.close();
    }

    public CuratorFramework getClient() {
        return client;
    }

    public String getLatchPath() {
        return latchPath;
    }

    public String getId() {
        return id;
    }

    public LeaderLatch getLeaderLatch() {
        return leaderLatch;
    }
}

现在在我的应用程序中，我正在使用这样的代码 -

public void method01() {
    ZookeeperClient zookClient = LeaderElectionExecutor.getInstance().getZookClient();
    if (zookClient.isLeader()) {
        // do something
    }
}

public void method02() {
    ZookeeperClient zookClient = LeaderElectionExecutor.getInstance().getZookClient();
    if (zookClient.isLeader()) {
        // do something
    }
}

问题陈述:-

在 Curator 库中 - 在启动闩锁后立即调用 isLeader() 将不起作用。领导者的选择需要时间。仅出于这个原因，我在我的 LeaderElectionExecutor 代码中添加了 1 分钟的 sleep 时间，该代码工作正常，但我想这不是正确的方法。

有更好的方法吗？牢记这一点，我需要一种方法来检查我是否是领导者然后执行这段代码。我不能在一个方法中完成所有事情，所以我需要从不同的类和方法调用 isLeader 方法来检查我是否是领导者，然后只执行这段代码。

我使用的是 Zookeeper 3.4.5 和 Curator 1.7.1 版本。

Answer 1

有一次我解决了一个和你的问题非常相似的问题。我就是这样做的。

首先，我让 Spring 管理我的对象。因此，我有一个可通过容器注入(inject)的 LeaderLatch。 LeadershipWatcher 是使用 LeaderLatch 的组件之一，它是 Runnable 接口(interface)的一种实现，可将领导事件分派(dispatch)给其他组件。最后这些组件是我命名为 LeadershipObserver 的接口(interface)的实现。 LeadershipWatcher 的实现主要类似于以下代码:

@Component
public class LeadershipWatcher implements Runnable {
  private final LeaderLatch leaderLatch;
  private final Collection<LeadershipObserver> leadershipObservers;

  /* constructor with @Inject */

  @Override
  public void run() {
    try {
      leaderLatch.await();

      for (LeadershipObserver observer : leadershipObservers) {
        observer.granted();
      }
    } catch (InterruptedException e) {
      for (LeadershipObserver observer : leadershipObservers) {
        observer.interrupted();
      }
    }
  }
}

因为这只是一个草图，我建议你增强这段代码，也许应用命令模式来调用观察者，或者甚至将观察者提交给线程池，如果他们的工作是阻塞的或长时间运行的 CPU 密集型任务。