Curator实现基于zookeeper leader选举---Spark学习技巧

一，基本介绍

Curator Framework是一个针对zookeeper做的搞层次的API，极大地简化了zookeeper的使用。它基于zookeeper构建了很多功能，处理复杂的链接管理，重试操作。下面是它的一些特点：

1，自动连接管理。

A),存在的潜在错误情况,需要ZooKeeper客户端创建连接和/或重试操作。Curator 自动和透明（主要）处理这些情况。

B),监视NodeDataChanged事件，并根据需要调用updateServerList（）。

C),Watches 会被Curator recipes自动移除。

2，简洁的API

A),提供现代简洁的接口

B),简化了zookeeper原生的方法，事件等。

3，实现了很多功能

A),Leader选举

B),共享锁

C),队列

D),Barriers

E),Counters

F),Caches

G),Nodes

本文主要是讲解leader选举。

现在分布式系统，基本架构模型架构是master/slaves，

二，leader选举

分布式系统中，一般都是master/slaves结构，为了容灾一般都是希望启动多个master，一个Leader，若干Followers，Leader宕机有Followers选举一个再做Leader。之所以，多个Master之间需要选举出Leader，实际上是为了保证数据的强一致性。

Zookeeper就是我们做Leader选举和配置下发的好的一种框架。

CuratorFramework

该类实例是线程安全的，一个应用程序最好公用一个。可以通过CuratorFrameworkFactory创建实例。

Leader选举的主要类，两个构造

public LeaderLatch(CuratorFramework client, String latchPath)Parameters: client - the client latchPath - the path for this leadership grouppublic LeaderLatch(CuratorFramework client, String latchPath, String id)Parameters: client - the client latchPath - the path for this leadership groupid - participant ID

构建了LeaderLatch对象后，需要对象调用start方法启动，我们然后就可以通过使用

public boolean hasLeadership()Return true if leadership is currently held by this instance

来判断当前实例是否是Leader。

本文采取的Leader选举的方案是spark的Master HA采用的方案<Spark源码之Standalone模式下master持久化引擎讲解>。也即是实现LeaderLatchListener进而通过它来监听动作master变化。

1，maven依赖

<properties> <curator.version>3.2.1</curator.version></properties><dependency> <groupId>org.apache.curator</groupId> <artifactId>curator-framework</artifactId> <version>${curator.version}</version></dependency> <!-- https://mvnrepository.com/artifact/org.apache.curator/curator-recipes --> <dependency> <groupId>org.apache.curator</groupId> <artifactId>curator-recipes</artifactId> <version>${curator.version}</version> <exclusions> <exclusion> <groupId>org.apache.zookeeper</groupId> <artifactId>zookeeper</artifactId> </exclusion> </exclusions> </dependency>

2，LeaderLatchListenerImpl

public class LeaderLatchListenerImpl implements LeaderLatchListener { @Override public void isLeader() { // TODO Auto-generated method stub System.out.println("master"); } @Override public void notLeader() { // TODO Auto-generated method stub System.out.println("slave"); }}

3，LeaderLatchListener

public class LeaderLatchMain { private static final String PATH = "/examples/leader"; public static void main(String[] args) throws Exception { CuratorFramework client = CuratorFrameworkFactory.newClient("106.75.13.87:2181", new ExponentialBackoffRetry(1000, 3)); client.start(); LeaderLatch leaderLatch = new LeaderLatch(client, PATH,args[0]); leaderLatch.addListener(new LeaderLatchListenerImpl()); try { leaderLatch.start(); Thread.sleep(10000); System.out.println(leaderLatch.getLeader()); Thread.sleep(Integer.MAX_VALUE); } catch (Exception e) { // TODO Auto-generated catch block e.printStackTrace(); leaderLatch.close(); client.close(); } }}

三，总结

本demo中participant ID是启动的时候当参数传递进去。

启动指令：

java -jar LeaderLatchMain.jar 1java -jar LeaderLatchMain.jar 2

第一次启动的时候leader

杀死Leader后

基于Curator还有一种leader 选举方式，下次介绍。 ---来自腾讯云社区的---Spark学习技巧