您现在的位置是:网站首页> 内容页

Zookeeper应用之——选举(Election)

  • ju111net免费影城
  • 2019-03-27
  • 149人已阅读
简介请注意,此篇文章并不是介绍Zookeeper集群内部Leader的选举机制,而是应用程序使用Zookeeper作为选举。使用Zookeeper进行选举,主要用到了Znode的两个性质:

请注意,此篇文章并不是介绍Zookeeper集群内部Leader的选举机制,而是应用程序使用Zookeeper作为选举。

使用Zookeeper进行选举,主要用到了Znode的两个性质:

    临时节点(EPHEMERAL)序列化节点(SEQUENCE)

每一个临时的序列化节点代表着一个客户端(client),也就是选民。主要的设计思路如下:

首先,创建一个选举的节点,我们叫做/election。 然后,每有一个客户端加入,就创建一个子节点/election/n_xxx,这个节点是EPHEMERAL并且SEQUENCE,xxx就是序列化产生的单调递增的数字。 在所有子节点中,序列数字做小的被选举成Leader。

上面的并不是重点,重点是Leader失败的检测,Leader失败后,一个新的客户端(client)将被选举成Leader。实现这个过程的一个最简单的方式是 所有的客户端(client)都监听Leader节点,一旦Leader节点消失,将通知所有的客户端(client)执行Leader选举过程,序列数字最小的将被选举成Leader。 这样实现看似没有问题,但是当客户端(client)数量非常庞大时,所有客户端(client)都将在/election节点执行getChildren(),这对Zookeeper 的压力是非常大的。为了避免这种“惊群效应”,我们可以让客户端只监听它前一个节点(所有序列数字比当前节点小,并且是其中最大的那个节点)。 这样,Leader节点消失后,哪个节点收到了通知,哪个节点就变成Leader,因为所有节点中,没有比它序列更小的节点了。

具体步骤如下:

    使用EPHEMERAL和SEQUENCE创建节点/election/n_xxx,我们叫做z。C为/election的子节点集合,i是z的序列数字。监听/election/n_j,j是C中小于i的最大数字。

接收到节点消失的事件后:

    C为新的/election的子节点集合如果z是集合中最小的节点,则z被选举成Leader如果z不是最小节点,则继续监听/election/n_j,j是C中小于i的最大数字。

具体代码如下:

public class Candidate implements Runnable, Watcher { //zk private ZooKeeper zk; //临时节点前缀 private String perfix = "n_"; //当前节点 private String currentNode; //前一个最大节点 private String lastNode; /** * 构造函数 * @param address zk地址 */ public Candidate(String address) { try { this.zk = new ZooKeeper(address, 3000, this); } catch (IOException e) { e.printStackTrace(); } } /** * 加入选举 */ @Override public void run() { try { //创建临时节点 currentNode = zk.create("/zookeeper/election/" + perfix, Thread.currentThread().getName().getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); //选举 election(); } catch (KeeperException e) { e.printStackTrace(); } catch (InterruptedException e) { e.printStackTrace(); } } /** * 从小到大排序临时节点 * @param children * @return */ private List<String> getSortedNode(List<String> children) { return children.stream().sorted(((o1, o2) -> { String sequence1 = o1.split(perfix)[1]; String sequence2 = o2.split(perfix)[1]; BigDecimal decimal1 = new BigDecimal(sequence1); BigDecimal decimal2 = new BigDecimal(sequence2); int result = decimal1.compareTo(decimal2); return result; })).collect(toList()); } /** * 选举过程 */ private void election(){ try{ while (true){ //获取/election节点中的所有子节点 List<String> children = zk.getChildren("/zookeeper/election", false); //所有子节点排序(从小到大) List<String> sortedNodes = getSortedNode(children); //获取最小节点 String smallestNode = sortedNodes.get(0); //当前节点就是最小节点,被选举成Leader if (currentNode.equals("/zookeeper/election/"+smallestNode)) { System.out.println(currentNode + "被选举成Leader。"); Thread.sleep(5000); //模拟Leader节点死去 System.out.println(currentNode+"已离去"); zk.close(); break; } //当前节点不是最小节点,监听前一个最大节点 else { //前一个最大节点 lastNode = smallestNode; //找到前一个最大节点,并监听 for (int i = 1; i < sortedNodes.size(); i++) { String z = sortedNodes.get(i); //找到前一个最大节点,并监听 if (currentNode.equals("/zookeeper/election/"+z)) { zk.exists("/zookeeper/election/" + lastNode, true); System.out.println(currentNode+"监听"+lastNode); //等待被唤起执行Leader选举 synchronized (this){ wait(); } break; } lastNode = z; } } } }catch (Exception e) { e.printStackTrace(); } } /** * 观察器通知 * @param event */ @Override public void process(WatchedEvent event) { //监听节点删除事件 if (event.getType().equals(Event.EventType.NodeDeleted)) { //被删除的节点是前一个最大节点,唤起线程执行选举 if (event.getPath().equals("/zookeeper/election/" + lastNode)) { System.out.println(currentNode+"被唤起"); synchronized (this){ notify(); } } } }}

我们将启动5个线程作为参选者,模拟每一个Leader死去,并重新选举的过程。启动程序如下:

public class Application { private static final String ADDRESS = "149.28.37.147:2181"; public static void main(String[] args) throws InterruptedException { setLog(); ExecutorService es = Executors.newFixedThreadPool(5); for (int i=0;i<5;i++){ es.execute(new Candidate(ADDRESS)); } es.shutdown(); } /** * 设置log级别为Error */ public static void setLog(){ //1.logback LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory(); //获取应用中的所有logger实例 List<Logger> loggerList = loggerContext.getLoggerList(); //遍历更改每个logger实例的级别,可以通过http请求传递参数进行动态配置 for (ch.qos.logback.classic.Logger logger:loggerList){ logger.setLevel(Level.toLevel("ERROR")); } }}

运行结果如下:

/zookeeper/election/n_0000000133被选举成Leader。/zookeeper/election/n_0000000134监听n_0000000133/zookeeper/election/n_0000000137监听n_0000000136/zookeeper/election/n_0000000135监听n_0000000134/zookeeper/election/n_0000000136监听n_0000000135/zookeeper/election/n_0000000133已离去/zookeeper/election/n_0000000134被唤起/zookeeper/election/n_0000000134被选举成Leader。/zookeeper/election/n_0000000134已离去/zookeeper/election/n_0000000135被唤起/zookeeper/election/n_0000000135被选举成Leader。/zookeeper/election/n_0000000135已离去/zookeeper/election/n_0000000136被唤起/zookeeper/election/n_0000000136被选举成Leader。/zookeeper/election/n_0000000136已离去/zookeeper/election/n_0000000137被唤起/zookeeper/election/n_0000000137被选举成Leader。/zookeeper/election/n_0000000137已离去

Zookeeper作为选举的应用就介绍完了,项目示例请参考:https://github.com/liubo-tech/zookeeper-application。

文章评论

Top