zookeeper——leader选举(curator)

本文主要介绍如何通过curator框架进行zookeeper的leader选举。
主要参考:
http://curator.apache.org/getting-started.html

好,下面上货。
leader或者说master一般是在分布式集群中执行某种特定功能的一台机器,比如系统中的报警、管理等等,这些只能够由某一台机器进行的工作,可以分配给master进行处理。那么如何在整个分布式集群中找到这个master,或者说选举出这个master呢?我们可以利用zookeeper实现,同时在通过curator框架简化编程。下面是一个简单的实现。

1、首先需要引入curator的maven依赖
<!-- https://mvnrepository.com/artifact/org.apache.curator/curator-framework -->
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-framework</artifactId>
            <version>4.0.0</version>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.zookeeper</groupId>
                    <artifactId>zookeeper</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>3.4.10</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>4.0.0</version>
        </dependency>

2、然后需要建立连接的通用代码
package com.xueyou.zkdemo.zkUtils;


import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;

public class CreateClient {
    public static CuratorFramework createSimple(String connectionString) {
        // these are reasonable arguments for the ExponentialBackoffRetry. The first
        // retry will wait 1 second - the second will wait up to 2 seconds - the
        // third will wait up to 4 seconds.
        ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3);

        // The simplest way to get a CuratorFramework instance. This will use default values.
        // The only required arguments are the connection string and the retry policy
        return CuratorFrameworkFactory.newClient(connectionString, retryPolicy);
    }

    public static CuratorFramework createWithOptions(String connectionString, RetryPolicy retryPolicy, int connectionTimeoutMs, int sessionTimeoutMs) {
        // using the CuratorFrameworkFactory.builder() gives fine grained control
        // over creation options. See the CuratorFrameworkFactory.Builder javadoc
        // details
        return CuratorFrameworkFactory.builder()
                .connectString(connectionString)
                .retryPolicy(retryPolicy)
                .connectionTimeoutMs(connectionTimeoutMs)
                .sessionTimeoutMs(sessionTimeoutMs)
                // etc. etc.
                .build();
    }
}

3、可能需要一些通用的函数,这里总结了一些,当然最初的出处来自github,这里是引用了一下:
package com.xueyou.zkdemo.zkUtils;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.api.CuratorListener;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import java.util.List;
import java.util.concurrent.atomic.AtomicReference;
public class CuratorZkClientBridge {
    private final CuratorFramework curator;
    private final AtomicReference<CuratorListener> listener = new AtomicReference<CuratorListener>(null);

    /**
     * @param curator Curator instance to bridge
     */
    public CuratorZkClientBridge(CuratorFramework curator)
    {
        this.curator = curator;
    }

    /**
     * Return the client
     *
     * @return client
     */
    public CuratorFramework getCurator()
    {
        return curator;
    }

    public void connect(final Watcher watcher)
    {
        if ( watcher != null )
        {
            CuratorListener     localListener = new CuratorListener()
            {
                public void eventReceived(CuratorFramework client, CuratorEvent event) throws Exception
                {
                    if ( event.getWatchedEvent() != null )
                    {
                        watcher.process(event.getWatchedEvent());
                    }
                }
            };
            curator.getCuratorListenable().addListener(localListener);
            listener.set(localListener);

            try
            {
                BackgroundCallback callback = new BackgroundCallback()
                {
                    public void processResult(CuratorFramework client, CuratorEvent event) throws Exception
                    {
                        WatchedEvent        fakeEvent = new WatchedEvent(Watcher.Event.EventType.None, curator.getZookeeperClient().isConnected() ? Watcher.Event.KeeperState.SyncConnected : Watcher.Event.KeeperState.Disconnected, null);
                        watcher.process(fakeEvent);
                    }
                };
                curator.checkExists().inBackground(callback).forPath("/foo");
            }
            catch ( Exception e )
            {
                throw new RuntimeException(e);
            }
        }
    }

    public void close() throws InterruptedException
    {
        // NOTE: the curator instance is NOT closed here

        CuratorListener localListener = listener.getAndSet(null);
        if ( localListener != null )
        {
            curator.getCuratorListenable().removeListener(localListener);
        }
    }

    public String create(String path, byte[] data, CreateMode mode) throws KeeperException, InterruptedException
    {
        try
        {
            return curator.create().withMode(mode).forPath(path, data);
        }
        catch ( Exception e )
        {
            adjustException(e);
        }
        return null;    // will never execute
    }

    public void delete(String path) throws InterruptedException, KeeperException
    {
        try
        {
            curator.delete().forPath(path);
        }
        catch ( Exception e )
        {
            adjustException(e);
        }
    }

    public boolean exists(String path, boolean watch) throws KeeperException, InterruptedException
    {
        try
        {
            return watch ? (curator.checkExists().watched().forPath(path) != null) : (curator.checkExists().forPath(path) != null);
        }
        catch ( Exception e )
        {
            adjustException(e);
        }
        return false;   // will never execute
    }

    public List<String> getChildren(String path, boolean watch) throws KeeperException, InterruptedException
    {
        try
        {
            return watch ? curator.getChildren().watched().forPath(path) : curator.getChildren().forPath(path);
        }
        catch ( Exception e )
        {
            adjustException(e);
        }
        return null;   // will never execute
    }

    public byte[] readData(String path, Stat stat, boolean watch) throws KeeperException, InterruptedException
    {
        try
        {
            if ( stat != null )
            {
                return watch ? curator.getData().storingStatIn(stat).watched().forPath(path) : curator.getData().storingStatIn(stat).forPath(path);
            }
            else
            {
                return watch ? curator.getData().watched().forPath(path) : curator.getData().forPath(path);
            }
        }
        catch ( Exception e )
        {
            adjustException(e);
        }
        return null;   // will never execute
    }

    public void writeData(String path, byte[] data, int expectedVersion) throws KeeperException, InterruptedException
    {
        writeDataReturnStat(path, data, expectedVersion);
    }

    public Stat writeDataReturnStat(String path, byte[] data, int expectedVersion) throws KeeperException, InterruptedException {
        try
        {
            curator.setData().withVersion(expectedVersion).forPath(path, data);
        }
        catch ( Exception e )
        {
            adjustException(e);
        }
        return null; // will never execute
    }

    public ZooKeeper.States getZookeeperState()
    {
        try
        {
            return curator.getZookeeperClient().getZooKeeper().getState();
        }
        catch ( Exception e )
        {
            throw new RuntimeException(e);
        }
    }

    public long getCreateTime(String path) throws KeeperException, InterruptedException
    {
        try
        {
            Stat            stat = curator.checkExists().forPath(path);
            return (stat != null) ? stat.getCtime() : 0;
        }
        catch ( Exception e )
        {
            adjustException(e);
        }
        return 0;
    }

    public String getServers()
    {
        return curator.getZookeeperClient().getCurrentConnectionString();
    }

    private void adjustException(Exception e) throws KeeperException, InterruptedException
    {
        if ( e instanceof KeeperException )
        {
            throw (KeeperException)e;
        }

        if ( e instanceof InterruptedException )
        {
            throw (InterruptedException)e;
        }

        throw new RuntimeException(e);
    }
}



4、主程序中进行master的选举
package com.xueyou.zkdemo;

import com.xueyou.zkdemo.zkUtils.CreateClient;
import com.xueyou.zkdemo.zkUtils.CuratorZkClientBridge;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.leader.LeaderSelector;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListener;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListenerAdapter;
import org.apache.zookeeper.KeeperException;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

/**
 * Hello world!
 */
public class App {
    public static String connectionString = "192.168.0.66:62181,192.168.0.66:62182,192.168.0.66:62183";
    public static List<String> res = new ArrayList<>();
    public static final int ID = 3;

    public static void main(String[] args) throws Exception {
        System.out.println("Hello World!");
        CuratorFramework curatorFramework = CreateClient.createSimple(connectionString);
        curatorFramework.start();
        //doSomething to zookeeper
        CuratorZkClientBridge curatorZkClientBridge = new CuratorZkClientBridge(curatorFramework);

//        System.out.println(getNode(curatorZkClientBridge, "/"));
        //master 选举
        LeaderSelectorListener listener = new LeaderSelectorListenerAdapter() {
            public void takeLeadership(CuratorFramework client) throws Exception {
                // this callback will get called when you are the leader
                // do whatever leader work you need to and only exit
                // this method when you want to relinquish leadership
                System.out.println("i am master" + "\t" + ID + "\t" + new SimpleDateFormat("HH:mm:ss").format(new Date()));
//                Thread.sleep(2000);
                Thread.sleep(Integer.MAX_VALUE);
            }
        };

        LeaderSelector selector = new LeaderSelector(curatorFramework, "/amaster", listener);
        selector.autoRequeue();  // not required, but this is behavior that you will probably expect
        selector.start();
        Thread.sleep(Integer.MAX_VALUE);
    }

    //递归获取所有zookeeper下的节点
    public static List<String> getNode(CuratorZkClientBridge curatorZkClientBridge, String parentNode) {
        try {
            List<String> tmpList = curatorZkClientBridge.getChildren(parentNode, false);
            for (String tmp : tmpList) {
                String childNode = parentNode.equals("/") ? parentNode + tmp : parentNode + "/" + tmp;
                res.add(childNode);
                getNode(curatorZkClientBridge, childNode);
            }
            return res;
        } catch (KeeperException e) {
            e.printStackTrace();
            return null;
        } catch (InterruptedException e) {
            e.printStackTrace();
            return null;
        }
    }

}

5、运行结果:
在运行时首先开启一个程序,然后能够看到程序打印出的内容:



然后同时修改ID的值,打开第二个程序,打开完成后发现并没有打印 i am master。这是关闭第一个程序,过一会能够看到第二个程序打印出下面的内容:


需要注意的是,这里的Thread.sleep(Integer.MAX_VALUE)是不能去掉的,因为去掉之后,马上回进行一下此的leader选举。当然,如果你需要leader不止一次执行一些代码,那么是可以去掉的。但是一般情况下,是通过master进行一个定时任务的执行,所以一般一次就可以了。
Logo

权威|前沿|技术|干货|国内首个API全生命周期开发者社区

更多推荐