【学习】026 Zookeeper

vi /etc/profile

source /etc/profile

cd /usr/local/zookeeper/conf
mv zoo_sample.cfg zoo.cfg

dataDir=/usr/local/zookeeper/data

server.0=bhz:2888:3888
server.1=hadoop1:2888:3888
server.2=hadoop2:2888:3888

#The number of milliseconds of each tick 
tickTime=2000 
#The number of ticks that the initial  
#synchronization phase can take 
initLimit=10 
#The number of ticks that can pass between  
#sending a request and getting an acknowledgement 
syncLimit=5 
#the directory where the snapshot is stored. 
#do not use /tmp for storage, /tmp here is just  
#example sakes. 
dataDir=/home/myuser/zooA/data 
#the port at which the clients will connect 
clientPort=2181 
#ZooKeeper server and its port no. # ZooKeeper ensemble should know about every other machine in the ensemble # specify server id by creating 'myid' file in the dataDir # use hostname instead of IP address for convenient maintenance
server.1=127.0.0.1:2888:3888
server.2=127.0.0.1:2988:3988
server.3=127.0.0.1:2088:3088 
# 
#Be sure to read the maintenance section of the  
#administrator guide before turning on autopurge. 
# 
#http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance 
# 
#The number of snapshots to retain in dataDir 
#autopurge.snapRetainCount=3 
#Purge task interval in hours 
#Set to "0" to disable auto purge feature  <br>
#autopurge.purgeInterval=1 
dataLogDir=/home/myuser/zooA/log

PERSISTENT                    #持久化节点
PERSISTENT_SEQUENTIAL        #顺序自动编号持久化节点，这种节点会根据当前已存在的节点数自动加 1
EPHEMERAL                     #临时节点， 客户端session超时这类节点就会被自动删除
EPHEMERAL_SEQUENTIAL         #临时自动编号节点

<dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>3.4.6</version>
        </dependency>

packagecom.hongmoshui.test;
importjava.io.IOException;
importjava.util.concurrent.CountDownLatch;
importorg.apache.zookeeper.CreateMode;
importorg.apache.zookeeper.KeeperException;
importorg.apache.zookeeper.WatchedEvent;
importorg.apache.zookeeper.Watcher;
importorg.apache.zookeeper.Watcher.Event.EventType;
importorg.apache.zookeeper.Watcher.Event.KeeperState;
importorg.apache.zookeeper.ZooDefs.Ids;
importorg.apache.zookeeper.ZooKeeper;
public classTest001
{
    //连接地址
    private static final String ADDRES = "127.0.0.1:2181";
    //session 会话
    private static final int SESSION_OUTTIME = 2000;
    //信号量,阻塞程序执行,用户等待zookeeper连接成功,发送成功信号，
    private static final CountDownLatch countDownLatch = new CountDownLatch(1);
    public static void main(String[] args) throwsIOException, InterruptedException, KeeperException
    {
        ZooKeeper zk = new ZooKeeper(ADDRES, SESSION_OUTTIME, newWatcher()
        {
            public voidprocess(WatchedEvent event)
            {
                //获取事件状态
                KeeperState keeperState =event.getState();
                //获取事件类型
                EventType eventType =event.getType();
                if (KeeperState.SyncConnected ==keeperState)
                {
                    if (EventType.None ==eventType)
                    {
                        countDownLatch.countDown();
                        System.out.println("zk 启动连接...");
                    }
                }
            }
        });
        //进行阻塞
countDownLatch.await();
        String result = zk.create("/hongmoshui_Lasting", "Lasting".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        System.out.println(result);
        zk.close();
    }
}

        //1. 创建持久节点，并且允许任何服务器可以操作
        String result = zk.create("/hongmoshui_Lasting", "Lasting".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        System.out.println("result:" +result);
        //2. 创建临时节点
        String result = zk.create("/hongmoshui_temp", "temp".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        System.out.println("result:" + result);

packagecom.hongmoshui;
importjava.util.concurrent.CountDownLatch;
importorg.apache.zookeeper.CreateMode;
importorg.apache.zookeeper.KeeperException;
importorg.apache.zookeeper.WatchedEvent;
importorg.apache.zookeeper.Watcher;
importorg.apache.zookeeper.Watcher.Event.EventType;
importorg.apache.zookeeper.Watcher.Event.KeeperState;
importorg.apache.zookeeper.ZooDefs.Ids;
importorg.apache.zookeeper.ZooKeeper;
importorg.apache.zookeeper.data.Stat;
public class ZkClientWatcher implementsWatcher
{
    //集群连接地址
    private static final String CONNECT_ADDRES = "192.168.110.159:2181,192.168.110.160:2181,192.168.110.162:2181";
    //会话超时时间
    private static final int SESSIONTIME = 2000;
    //信号量,让zk在连接之前等待,连接成功后才能往下走.
    private static final CountDownLatch countDownLatch = new CountDownLatch(1);
    private static String LOG_MAIN = "【main】 ";
    privateZooKeeper zk;
    public void createConnection(String connectAddres, intsessionTimeOut)
    {
        try
        {
            zk = new ZooKeeper(connectAddres, sessionTimeOut, this);
            System.out.println(LOG_MAIN + "zk 开始启动连接服务器....");
            countDownLatch.await();
        }
        catch(Exception e)
        {
            e.printStackTrace();
        }
    }
    public booleancreatePath(String path, String data)
    {
        try
        {
            this.exists(path, true);
            this.zk.create(path, data.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            System.out.println(LOG_MAIN + "节点创建成功, Path:" + path + ",data:" +data);
        }
        catch(Exception e)
        {
            e.printStackTrace();
            return false;
        }
        return true;
    }
    /**
     * 判断指定节点是否存在
     * 
     * @parampath 节点路径
     */
    public Stat exists(String path, booleanneedWatch)
    {
        try
        {
            return this.zk.exists(path, needWatch);
        }
        catch(Exception e)
        {
            e.printStackTrace();
            return null;
        }
    }
    public boolean updateNode(String path, String data) throwsKeeperException, InterruptedException
    {
        exists(path, true);
        this.zk.setData(path, data.getBytes(), -1);
        return false;
    }
    public voidprocess(WatchedEvent watchedEvent)
    {
        //获取事件状态
        KeeperState keeperState =watchedEvent.getState();
        //获取事件类型
        EventType eventType =watchedEvent.getType();
        //zk 路径
        String path =watchedEvent.getPath();
        System.out.println("进入到 process() keeperState:" + keeperState + ", eventType:" + eventType + ", path:" +path);
        //判断是否建立连接
        if (KeeperState.SyncConnected ==keeperState)
        {
            if (EventType.None ==eventType)
            {
                //如果建立建立成功,让后程序往下走
                System.out.println(LOG_MAIN + "zk 建立连接成功!");
                countDownLatch.countDown();
            }
            else if (EventType.NodeCreated ==eventType)
            {
                System.out.println(LOG_MAIN + "事件通知,新增node节点" +path);
            }
            else if (EventType.NodeDataChanged ==eventType)
            {
                System.out.println(LOG_MAIN + "事件通知,当前node节点" + path + "被修改....");
            }
            else if (EventType.NodeDeleted ==eventType)
            {
                System.out.println(LOG_MAIN + "事件通知,当前node节点" + path + "被删除....");
            }
        }
        System.out.println("--------------------------------------------------------");
    }
    public static void main(String[] args) throwsKeeperException, InterruptedException
    {
        ZkClientWatcher zkClientWatcher = newZkClientWatcher();
        zkClientWatcher.createConnection(CONNECT_ADDRES, SESSIONTIME);
//boolean createResult = zkClientWatcher.createPath("/p15", "pa-644064");
        zkClientWatcher.updateNode("/pa2", "7894561");
    }
}

packagecom.hongmoshui.distributed;
importjava.text.SimpleDateFormat;
importjava.util.Date;
//生成订单类
public classOrderNumGenerator
{
    //全局订单id
    public static int count = 0;
    publicString getNumber()
    {
        try
        {
            Thread.sleep(200);
        }
        catch(Exception e)
        {
        }
        SimpleDateFormat simpt = new SimpleDateFormat("yyyy-MM-dd-HH-mm-ss");
        return simpt.format(new Date()) + "-" + ++count;
    }
}

packagecom.hongmoshui.distributed;
//使用多线程模拟生成订单号
public class OrderService implementsRunnable
{
    private OrderNumGenerator orderNumGenerator = newOrderNumGenerator();
    public voidrun()
    {
        getNumber();
    }
    public voidgetNumber()
    {
        String number =orderNumGenerator.getNumber();
        System.out.println(Thread.currentThread().getName() + ",生成订单ID:" +number);
    }
    public static voidmain(String[] args)
    {
        System.out.println("####生成唯一订单号###");
        for (int i = 0; i < 100; i++)
        {
            new Thread(newOrderService()).start();
        }
    }
}

packagecom.hongmoshui.distributed;
//使用多线程模拟生成订单号
public class OrderSynchronizedService implementsRunnable
{
    private OrderNumGenerator orderNumGenerator = newOrderNumGenerator();
    public voidrun()
    {
        getNumber();
    }
    public voidgetNumber()
    {
        synchronized (this)
        {
            String number =orderNumGenerator.getNumber();
            System.out.println(Thread.currentThread().getName() + ",生成订单ID:" +number);
        }
    }
    public static voidmain(String[] args)
    {
        System.out.println("####生成唯一订单号###");
        OrderService orderService = newOrderService();
        for (int i = 0; i < 100; i++)
        {
            newThread(orderService).start();
        }
    }
}

packagecom.hongmoshui.distributed;
importjava.util.concurrent.locks.ReentrantLock;
public class OrderLockService implementsRunnable
{
    private OrderNumGenerator orderNumGenerator = newOrderNumGenerator();
    //使用lock锁
    private java.util.concurrent.locks.Lock lock = newReentrantLock();
    public voidrun()
    {
        getNumber();
    }
    public voidgetNumber()
    {
        try
        {
            //synchronized (this) {
lock.lock();
            String number =orderNumGenerator.getNumber();
            System.out.println(Thread.currentThread().getName() + ",生成订单ID:" +number);
            //}

        }
        catch(Exception e)
        {
        }
        finally
        {
            lock.unlock();
        }
    }
    public static voidmain(String[] args)
    {
        System.out.println("####生成唯一订单号###");
        OrderService orderService = newOrderService();
        for (int i = 0; i < 100; i++)
        {
            newThread(orderService).start();
        }
    }
}

        <dependency>
            <groupId>com.101tec</groupId>
            <artifactId>zkclient</artifactId>
            <version>0.10</version>
        </dependency>

packagecom.hongmoshui.distributed;
public interfaceLock
{
    //获取到锁的资源
    public voidgetLock();
    //释放锁
    public voidunLock();
}

packagecom.hongmoshui.distributed;
importorg.I0Itec.zkclient.ZkClient;
//将重复代码写入子类中..
public abstract class ZookeeperAbstractLock implementsLock
{
    //zk连接地址
    private static final String CONNECTSTRING = "127.0.0.1:2181";
    //创建zk连接
    protected ZkClient zkClient = newZkClient(CONNECTSTRING);
    protected static final String PATH = "/lock";
    public voidgetLock()
    {
        if(tryLock())
        {
            System.out.println("##获取lock锁的资源####");
        }
        else
        {
            //等待
waitLock();
            //重新获取锁资源
getLock();
        }
    }
    //获取锁资源
    abstract booleantryLock();
    //等待
    abstract voidwaitLock();
    public voidunLock()
    {
        if (zkClient != null)
        {
            zkClient.close();
            System.out.println("释放锁资源...");
        }
    }
}

packagecom.hongmoshui.distributed;
importjava.util.concurrent.CountDownLatch;
importorg.I0Itec.zkclient.IZkDataListener;
public class ZookeeperDistrbuteLock extendsZookeeperAbstractLock
{
    private CountDownLatch countDownLatch = null;
    @Override
    booleantryLock()
    {
        try
        {
            zkClient.createEphemeral(PATH);
            return true;
        }
        catch(Exception e)
        {
//e.printStackTrace();
            return false;
        }
    }
    @Override
    voidwaitLock()
    {
        IZkDataListener izkDataListener = newIZkDataListener()
        {
            public void handleDataDeleted(String path) throwsException
            {
                //唤醒被等待的线程
                if (countDownLatch != null)
                {
                    countDownLatch.countDown();
                }
            }
            public void handleDataChange(String path, Object data) throwsException
            {
            }
        };
        //注册事件
zkClient.subscribeDataChanges(PATH, izkDataListener);
        if(zkClient.exists(PATH))
        {
            countDownLatch = new CountDownLatch(1);
            try
            {
                countDownLatch.await();
            }
            catch(Exception e)
            {
                e.printStackTrace();
            }
        }
        //删除监听
zkClient.unsubscribeDataChanges(PATH, izkDataListener);
    }
}

packagecom.hongmoshui.distributed;
importcom.hongmoshui.OrderNumGenerator;
public class OrderService implementsRunnable
{
    private OrderNumGenerator orderNumGenerator = newOrderNumGenerator();
    //使用lock锁
    //private
    //java.util.concurrent.locks.Lock
    //lock = new ReentrantLock();
    private Lock lock = newZookeeperDistrbuteLock();
    public voidrun()
    {
        getNumber();
    }
    public voidgetNumber()
    {
        try
        {
            lock.getLock();
            String number =orderNumGenerator.getNumber();
            System.out.println(Thread.currentThread().getName() + ",生成订单ID:" +number);
        }
        catch(Exception e)
        {
            e.printStackTrace();
        }
        finally
        {
            lock.unLock();
        }
    }
    public static voidmain(String[] args)
    {
        System.out.println("####生成唯一订单号###");
//OrderService orderService = new OrderService();
        for (int i = 0; i < 100; i++)
        {
            new Thread(newOrderService()).start();
        }
    }
}

        <dependency>
            <groupId>com.101tec</groupId>
            <artifactId>zkclient</artifactId>
            <version>0.8</version>
        </dependency>

packagecom.hongmoshui.LoadBalance;
importjava.io.BufferedReader;
importjava.io.IOException;
importjava.io.InputStreamReader;
importjava.io.PrintWriter;
importjava.net.Socket;
//ServerHandler
public class ServerHandler implementsRunnable
{
    privateSocket socket;
    publicServerHandler(Socket socket)
    {
        this.socket =socket;
    }
    public voidrun()
    {
        BufferedReader in = null;
        PrintWriter out = null;
        try
        {
            in = new BufferedReader(new InputStreamReader(this.socket.getInputStream()));
            out = new PrintWriter(this.socket.getOutputStream(), true);
            String body = null;
            while (true)
            {
                body =in.readLine();
                if (body == null)
                    break;
                System.out.println("Receive : " +body);
                out.println("Hello, " +body);
            }
        }
        catch(Exception e)
        {
            if (in != null)
            {
                try
                {
                    in.close();
                }
                catch(IOException e1)
                {
                    e1.printStackTrace();
                }
            }
            if (out != null)
            {
                out.close();
            }
            if (this.socket != null)
            {
                try
                {
                    this.socket.close();
                }
                catch(IOException e1)
                {
                    e1.printStackTrace();
                }
                this.socket = null;
            }
        }
    }
}

packagecom.hongmoshui.LoadBalance;
importjava.io.IOException;
importjava.net.ServerSocket;
importjava.net.Socket;
//##ServerScoekt服务端
public class ZkServerScoekt implementsRunnable
{
    private int port = 18080;
    public static void main(String[] args) throwsIOException
    {
        int port = 18080;
        ZkServerScoekt server = newZkServerScoekt(port);
        Thread thread = newThread(server);
        thread.start();
    }
    public ZkServerScoekt(intport)
    {
        this.port =port;
    }
    public voidrun()
    {
        ServerSocket serverSocket = null;
        try
        {
            serverSocket = newServerSocket(port);
            System.out.println("Server start port:" +port);
            Socket socket = null;
            while (true)
            {
                socket =serverSocket.accept();
                new Thread(newServerHandler(socket)).start();
            }
        }
        catch(Exception e)
        {
            e.printStackTrace();
        }
        finally
        {
            try
            {
                if (serverSocket != null)
                {
                    serverSocket.close();
                }
            }
            catch(Exception e2)
            {
            }
        }
    }
}

packagecom.hongmoshui.LoadBalance;
importjava.io.BufferedReader;
importjava.io.IOException;
importjava.io.InputStreamReader;
importjava.io.PrintWriter;
importjava.net.Socket;
importjava.util.ArrayList;
importjava.util.List;
public classZkServerClient
{
    public static List<String> listServer = new ArrayList<String>();
    public static voidmain(String[] args)
    {
        initServer();
        ZkServerClient client = newZkServerClient();
        BufferedReader console = new BufferedReader(newInputStreamReader(System.in));
        while (true)
        {
            String name;
            try
            {
                name =console.readLine();
                if ("exit".equals(name))
                {
                    System.exit(0);
                }
                client.send(name);
            }
            catch(IOException e)
            {
                e.printStackTrace();
            }
        }
    }
    //注册所有server
    public static voidinitServer()
    {
        listServer.clear();
        listServer.add("127.0.0.1:18080");
    }
    //获取当前server信息
    public staticString getServer()
    {
        return listServer.get(0);
    }
    public voidsend(String name)
    {
        String server =ZkServerClient.getServer();
        String[] cfg = server.split(":");
        Socket socket = null;
        BufferedReader in = null;
        PrintWriter out = null;
        try
        {
            socket = new Socket(cfg[0], Integer.parseInt(cfg[1]));
            in = new BufferedReader(newInputStreamReader(socket.getInputStream()));
            out = new PrintWriter(socket.getOutputStream(), true);
            out.println(name);
            while (true)
            {
                String resp =in.readLine();
                if (resp == null)
                    break;
                else if (resp.length() > 0)
                {
                    System.out.println("Receive : " +resp);
                    break;
                }
            }
        }
        catch(Exception e)
        {
            e.printStackTrace();
        }
        finally
        {
            if (out != null)
            {
                out.close();
            }
            if (in != null)
            {
                try
                {
                    in.close();
                }
                catch(IOException e)
                {
                    e.printStackTrace();
                }
            }
            if (socket != null)
            {
                try
                {
                    socket.close();
                }
                catch(IOException e)
                {
                    e.printStackTrace();
                }
            }
        }
    }
}

packagecom.hongmoshui.LoadBalance;
importjava.io.IOException;
importjava.net.ServerSocket;
importjava.net.Socket;
importorg.I0Itec.zkclient.ZkClient;
public class ZkServerScoekt2 implementsRunnable
{
    private static int port = 18081;
    public static void main(String[] args) throwsIOException
    {
        ZkServerScoekt server = newZkServerScoekt(port);
        Thread thread = newThread(server);
        thread.start();
    }
    public ZkServerScoekt2(intport)
    {
        this.port =port;
    }
    public voidregServer()
    {
        //向ZooKeeper注册当前服务器
        ZkClient client = new ZkClient("127.0.0.1:2181", 60000, 1000);
        String path = "/test/server" +port;
        if(client.exists(path))
            client.delete(path);
        client.createEphemeral(path, "127.0.0.1:" +port);
    }
    public voidrun()
    {
        ServerSocket serverSocket = null;
        try
        {
            serverSocket = newServerSocket(port);
            regServer();
            System.out.println("Server start port:" +port);
            Socket socket = null;
            while (true)
            {
                socket =serverSocket.accept();
                new Thread(newServerHandler(socket)).start();
            }
        }
        catch(Exception e)
        {
            e.printStackTrace();
        }
        finally
        {
            try
            {
                if (serverSocket != null)
                {
                    serverSocket.close();
                }
            }
            catch(Exception e2)
            {
            }
        }
    }
}

packagecom.hongmoshui.LoadBalance;
importjava.io.BufferedReader;
importjava.io.IOException;
importjava.io.InputStreamReader;
importjava.io.PrintWriter;
importjava.net.Socket;
importjava.util.ArrayList;
importjava.util.List;
importorg.I0Itec.zkclient.IZkChildListener;
importorg.I0Itec.zkclient.ZkClient;
public classZkServerClient2
{
    public static List<String> listServer = new ArrayList<String>();
    public static voidmain(String[] args)
    {
        initServer();
        ZkServerClient client = newZkServerClient();
        BufferedReader console = new BufferedReader(newInputStreamReader(System.in));
        while (true)
        {
            String name;
            try
            {
                name =console.readLine();
                if ("exit".equals(name))
                {
                    System.exit(0);
                }
                client.send(name);
            }
            catch(IOException e)
            {
                e.printStackTrace();
            }
        }
    }
    //注册所有server
    public static voidinitServer()
    {
        final String path = "/test";
        final ZkClient zkClient = new ZkClient("127.0.0.1:2181", 60000, 1000);
        List<String> children =zkClient.getChildren(path);
        listServer.clear();
        for(String p : children)
        {
            listServer.add((String) zkClient.readData(path + "/" +p));
        }
        //订阅节点变化事件
        zkClient.subscribeChildChanges("/test", newIZkChildListener()
        {
            public void handleChildChange(String parentPath, List<String> currentChilds) throwsException
            {
                listServer.clear();
                for(String p : currentChilds)
                {
                    listServer.add((String) zkClient.readData(path + "/" +p));
                }
                System.out.println("####handleChildChange()####listServer:" +listServer.toString());
            }
        });
    }
    //请求次数
    private static int count = 1;
    //服务数量
    private static int serverCount = 2;
    //获取当前server信息
    public staticString getServer()
    {
        String serverName = listServer.get(count %serverCount);
        ++count;
        returnserverName;
    }
    public voidsend(String name)
    {
        String server =ZkServerClient.getServer();
        String[] cfg = server.split(":");
        Socket socket = null;
        BufferedReader in = null;
        PrintWriter out = null;
        try
        {
            socket = new Socket(cfg[0], Integer.parseInt(cfg[1]));
            in = new BufferedReader(newInputStreamReader(socket.getInputStream()));
            out = new PrintWriter(socket.getOutputStream(), true);
            out.println(name);
            while (true)
            {
                String resp =in.readLine();
                if (resp == null)
                    break;
                else if (resp.length() > 0)
                {
                    System.out.println("Receive : " +resp);
                    break;
                }
            }
        }
        catch(Exception e)
        {
            e.printStackTrace();
        }
        finally
        {
            if (out != null)
            {
                out.close();
            }
            if (in != null)
            {
                try
                {
                    in.close();
                }
                catch(IOException e)
                {
                    e.printStackTrace();
                }
            }
            if (socket != null)
            {
                try
                {
                    socket.close();
                }
                catch(IOException e)
                {
                    e.printStackTrace();
                }
            }
        }
    }
}