zookeeper——leader选举（curator）

分手后的思念是犯贱 2022-06-07 02:49 262阅读 0赞

# zookeeper——leader选举（curator） #

本文主要介绍如何通过curator框架进行zookeeper的leader选举。

主要参考：

http://curator.apache.org/getting-started.html

好，下面上货。

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

1、首先需要引入curator的maven依赖

<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、运行结果：

在运行时首先开启一个程序，然后能够看到程序打印出的内容：

![Center][]

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

![Center 1][]

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

[Center]: /images/20220607/65a861252d124f399b94e9fc64b64b19.png
[Center 1]: /images/20220607/71c23632c0904c598a4dffc3fe3f7e5e.png