参考：
官文文档：http://curator.apache.org/curator-recipes/leader-latch.html
http://curator.apache.org/curator-recipes/leader-election.html
技术博客： http://blog.csdn.net/smallnest/article/details/41895431

分布式计算中，选举就是指定一个线程用来选择多个节点的领导者。在任务开始时，所有的节点都不知道哪个为领导者。选举算法开始后，领导者选出来后，每个节点通过网络知道领导者的存在。
Curator中选举分为两种： Leader Latch和Leader Election

Leader Latch

很简单的选举算法。随机从候选者中选择一台作为leader，选中后除非leader自己调用close（）释放leadership，否则其他的候选者不能成为leader

使用方法

初始化LeaderLatch

public LeaderLatch(CuratorFramework client,String latchPath)
Parameters:
client - the client
latchPath - the path for this leadership group

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

启动

通过start()方法启动

leaderLatch.start();

一旦启动，LeaderLatch会与其他有相同latchpath的候选者协商，从中随机选中一个作为leader. 可以调用实例的hasLeadership()判断该实例是否为leader。

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

尝试获取leadership

LeaderLatch与JDK的CountDownLatch类似，调用await（）方法会使线程一直阻塞到获得leadership为止。

public void await()throws InterruptedException,EOFException
Causes the current thread to wait until this instance acquires leadership
unless the thread is interrupted or closed.

public boolean await(long timeout,TimeUnit unit)throws InterruptedException
Causes the current thread to wait until this instance acquires leadership unless
the thread is interrupted, the specified waiting time elapses or the instance is closed.
&nbsp;
Parameters:
timeout - the maximum time to wait
unit - the time unit of the timeout argument
Returns:
true if the count reached zero and false if the waiting time elapsed before the count
reached zero or the instances was closed

释放leadership

只能通过close()释放leadership, 只有leader将leadership释放时，其他的候选者才有机会被选为leader

leaderLatch.close();

错误处理

LeaderLatch通过增加了一个ConnectionStateListener监听连接问题。如果出现SUSPENDED或者LOST，leader会报告自己不再是leader（直到重新建立连接，否则不会有leader）。如果LOST的连接被重新建立即RECONNECTED，leaderLatch会删除先前的zNode并重新建立zNode。
强烈建议LeaderLatch使用时注册一个ConnectionStateListener。

实例

添加依赖

        <dependency><groupId>org.apache.curator</groupId><artifactId>curator-recipes</artifactId><version>2.6.0</version></dependency><dependency><groupId>org.apache.curator</groupId><artifactId>curator-test</artifactId><version>2.6.0</version></dependency>

代码如下：

package curator;import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.List;
import java.util.concurrent.TimeUnit;import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.test.TestingServer;
import org.apache.curator.utils.CloseableUtils;import com.google.common.collect.Lists;public class LeaderLatchExample {private static final int CLIENT_QTY=10;private static final String PATH="/examples/leader";
public static void main(String[] args){List<CuratorFramework> clients=Lists.newArrayList();List<LeaderLatch> examples=Lists.newArrayList();TestingServer server = null;try {server=new TestingServer();//创建了10个LeaderLatch，启动后它们中的一个会被选举为leaderfor(int i=0;i<CLIENT_QTY;i++){CuratorFramework client=CuratorFrameworkFactory.newClient(server.getConnectString(),new ExponentialBackoffRetry(1000,3));clients.add(client);LeaderLatch example=new LeaderLatch(client,PATH,"client #"+i);examples.add(example);client.start();example.start();}// 因为选举会花费一些时间，start后并不能马上就得到leader。Thread.sleep(20000);LeaderLatch currentLeader=null;for(int i=0;i<CLIENT_QTY;i++){LeaderLatch example=examples.get(i);//通过hasLeadership查看自己是否是leaderif(example.hasLeadership()){currentLeader=example;break;}}System.out.println("current leader is "+currentLeader.getId());System.out.println("release the leader "+currentLeader.getId());//只能通过close释放当前的领导权。currentLeader.close();//await是一个阻塞方法， 尝试获取leader地位，但是未必能上位。examples.get(0).await(2,TimeUnit.SECONDS);System.out.println("client #0 maybe is elected as the leader or not although it want to be");//可以通过.getLeader().getId()可以得到当前的leader的IDSystem.out.println("the new leader is "+examples.get(0).getLeader().getId());System.out.println("Press enter/return to quit \n");new BufferedReader(new InputStreamReader(System.in)).readLine();} catch (Exception e) {e.printStackTrace();}finally{System.out.println("Shutting down....");for(LeaderLatch exampleClient: examples){CloseableUtils.closeQuietly(exampleClient);}for(CuratorFramework client:clients){CloseableUtils.closeQuietly(client);}CloseableUtils.closeQuietly(server);}}
}

首先创建了10个LeaderLatch，启动后它们中的一个会被选举为leader。因为选举会花费一些时间，start后并不能马上就得到leader。
通过hasLeadership查看自己是否是leader，如果是的话返回true。
可以通过.getLeader().getId()可以得到当前的leader的ID。
只能通过close释放当前的领导权。
await是一个阻塞方法，尝试获取leader地位，但是未必能上位。
参考： http://blog.csdn.net/smallnest/article/details/41895431

Leader Election

通过LeaderSelectorListener可以对领导权进行控制，在适当的时候释放领导权，这样每个节点都有可能获得领导权。而LeaderLatch一根筋到死，除非调用close方法，否则它不会释放领导权。

使用方法

创建 LeaderSelector

public LeaderSelector(CuratorFramework client,String mutexPath,LeaderSelectorListener listener)
Parameters:
client - the client
mutexPath - the path for this leadership group
listener - listener

public LeaderSelector(CuratorFramework client,String mutexPath,ThreadFactory threadFactory,Executor executor,LeaderSelectorListener listener)
Parameters:
client - the client
mutexPath - the path for this leadership group
threadFactory - factory to use for making internal threads
executor - the executor to run in
listener - listener

启动

leaderSelector.start();

一旦启动，如果获取了leadership的话，takeLeadership（）会被调用，只有当leader释放了leadership的时候，takeLeadership（）才会返回。

释放

调用close（）释放 leadership

leaderSelector.close();

错误处理

LeaderSelectorListener类继承了ConnectionStateListener。一旦LeaderSelector启动，它会向curator客户端添加监听器。使用LeaderSelector必须时刻注意连接的变化。一旦出现连接问题如SUSPENDED，curator实例必须确保它可能不再是leader，直至它重新收到RECONNECTED。如果LOST出现，curator实例不再是leader并且其takeLeadership()应该直接退出。
建议重要：推荐的做法是，如果发生SUSPENDED或者LOST连接问题，最好直接抛CancelLeadershipException，此时，leaderSelector实例会尝试中断并且取消正在执行takeLeadership（）方法的线程。建议扩展LeaderSelectorListenerAdapter, LeaderSelectorListenerAdapter中已经提供了推荐的处理方式。

实例

来自于官方

/*** Licensed to the Apache Software Foundation (ASF) under one* or more contributor license agreements.  See the NOTICE file* distributed with this work for additional information* regarding copyright ownership.  The ASF licenses this file* to you under the Apache License, Version 2.0 (the* "License"); you may not use this file except in compliance* with the License.  You may obtain a copy of the License at**   http://www.apache.org/licenses/LICENSE-2.0** Unless required by applicable law or agreed to in writing,* software distributed under the License is distributed on an* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY* KIND, either express or implied.  See the License for the* specific language governing permissions and limitations* under the License.*/
package leader;import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListenerAdapter;
import org.apache.curator.framework.recipes.leader.LeaderSelector;
import java.io.Closeable;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;/*** An example leader selector client. Note that {@link LeaderSelectorListenerAdapter} which* has the recommended handling for connection state issues*/
public class ExampleClient extends LeaderSelectorListenerAdapter implements Closeable
{private final String name;private final LeaderSelector leaderSelector;private final AtomicInteger leaderCount = new AtomicInteger();public ExampleClient(CuratorFramework client, String path, String name){this.name = name;// create a leader selector using the given path for management// all participants in a given leader selection must use the same path// ExampleClient here is also a LeaderSelectorListener but this isn't requiredleaderSelector = new LeaderSelector(client, path, this);// for most cases you will want your instance to requeue when it relinquishes leadership//保证此实例在释放领导权后还可能获得领导权leaderSelector.autoRequeue();}public void start() throws IOException{// the selection for this instance doesn't start until the leader selector is started// leader selection is done in the background so this call to leaderSelector.start() returns immediatelyleaderSelector.start();}@Overridepublic void close() throws IOException{leaderSelector.close();}@Overridepublic void takeLeadership(CuratorFramework client) throws Exception{// we are now the leader. This method should not return until we want to relinquish leadershipfinal int         waitSeconds = (int)(5 * Math.random()) + 1;System.out.println(name + " is now the leader. Waiting " + waitSeconds + " seconds...");System.out.println(name + " has been leader " + leaderCount.getAndIncrement() + " time(s) before.");try{Thread.sleep(TimeUnit.SECONDS.toMillis(waitSeconds));}catch ( InterruptedException e ){System.err.println(name + " was interrupted.");Thread.currentThread().interrupt();}finally{System.out.println(name + " relinquishing leadership.\n");}}
}

其中： leaderSelector.autoRequeue()能保证该实例在释放leadership后还可能重新获得leadership。
成为leader时，会调用 takeLeadership（）方法，可以在这里面进行任务分配，并且不要返回，如果想要实例一直是leader可以加一个死循环。
测试代码：来自于官方

/*** Licensed to the Apache Software Foundation (ASF) under one* or more contributor license agreements.  See the NOTICE file* distributed with this work for additional information* regarding copyright ownership.  The ASF licenses this file* to you under the Apache License, Version 2.0 (the* "License"); you may not use this file except in compliance* with the License.  You may obtain a copy of the License at**   http://www.apache.org/licenses/LICENSE-2.0** Unless required by applicable law or agreed to in writing,* software distributed under the License is distributed on an* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY* KIND, either express or implied.  See the License for the* specific language governing permissions and limitations* under the License.*/
package leader;import com.google.common.collect.Lists;
import org.apache.curator.utils.CloseableUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.test.TestingServer;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.List;public class LeaderSelectorExample
{private static final int        CLIENT_QTY = 10;private static final String     PATH = "/examples/leader";public static void main(String[] args) throws Exception{// all of the useful sample code is in ExampleClient.javaSystem.out.println("Create " + CLIENT_QTY + " clients, have each negotiate for leadership and then wait a random number of seconds before letting another leader election occur.");System.out.println("Notice that leader election is fair: all clients will become leader and will do so the same number of times.");List<CuratorFramework>  clients = Lists.newArrayList();List<ExampleClient>     examples = Lists.newArrayList();TestingServer           server = new TestingServer();try{for ( int i = 0; i < CLIENT_QTY; ++i ){CuratorFramework    client = CuratorFrameworkFactory.newClient(server.getConnectString(), new ExponentialBackoffRetry(1000, 3));clients.add(client);ExampleClient       example = new ExampleClient(client, PATH, "Client #" + i);examples.add(example);client.start();example.start();}System.out.println("Press enter/return to quit\n");new BufferedReader(new InputStreamReader(System.in)).readLine();}finally{System.out.println("Shutting down...");for ( ExampleClient exampleClient : examples ){CloseableUtils.closeQuietly(exampleClient);}for ( CuratorFramework client : clients ){CloseableUtils.closeQuietly(client);}CloseableUtils.closeQuietly(server);}}
}

创建了10个客户端，通过公平选举成为leader。

Curator之Recipes之选举相关推荐

五、curator recipes之选举主节点Leader Latch
简介在分布式计算中,主节点选举是为了把某个进程作为主节点来控制其它节点的过程.在选举结束之前,我们不知道哪个节点会成为主节点.curator对于主节点选举有两种实现方式,本文示例演示Latch的实现 ...
ZooKeeper ： Curator框架之Leader选举LeaderLatch
Leader选举在分布式计算中,Leader选举是指指定单个进程作为分布在多台计算机(节点)之间的某些任务的组织者的过程.在任务开始之前,所有网络节点都不知道哪个节点将充当任务的Leader.然而, ...
springboot使用curator来实现leader选举
本文来说下springboot使用curator来实现leader选举文章目录概述概述
基于curator实现leader选举
1.前言 curator由Netflix的工程师开发,主要目的为了基于zookeeper的应用变得简单可靠,在2013年成为apache的顶级项目.curator基于zookeeper,但提供了更高级 ...
十六、curator recipes之DistributedIdQueue
简介 curator实现了一种分布式ID队列,也是遵循FIFO原则,比普通队列新增的一个点是ID队列可以根据ID对队列元素进行操作,比如移除该元素. 官方文档:http://curator.apach ...
Zookeeper系列五：Master选举、ZK高级特性：基本模型
一.Master选举 1. master选举原理: 有多个master,每次只能有一个master负责主要的工作,其他的master作为备份,同时对负责工作的master进行监听,一旦负责工作的mas ...
Curator selector
2019独角兽企业重金招聘Python工程师标准>>> zookeeper最经典的用法,选举.Curator的实现有2个种: LeaderLatch: 这种是有阻塞的,就是大家一起上 ...
zookeeper客户端库curator分析
zookeeper客户端库curator分析前言综述 zookeeper保证理解zookeeper的顺序一致性之前使用zookeeper客户端踩到的坑 curator 连接保证连接状态监控以 ...
Curator的基本使用
一.简介 Curator是Netflix公司开源的一套Zookeeper客户端框架.了解过Zookeeper原生API都会清楚其复杂度.Curator帮助我们在其基础上进行封装.实现一些开发细节,包括 ...

Curator之Recipes之选举

Leader Latch

相关的类

使用方法

初始化LeaderLatch

启动

尝试获取leadership

释放leadership

错误处理

实例

Leader Election

相关的类

使用方法

创建 LeaderSelector

启动

释放

错误处理

实例

Curator之Recipes之选举相关推荐

最新文章

热门文章