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>

package com.hongmoshui.test;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;

public class Test001
{

    // 连接地址
    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) throws IOException, InterruptedException, KeeperException
    {
        ZooKeeper zk = new ZooKeeper(ADDRES, SESSION_OUTTIME, new Watcher()
        {

            public void process(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);

package com.hongmoshui;
import java.util.concurrent.CountDownLatch;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

public class ZkClientWatcher implements Watcher
{
    // 集群连接地址
    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】 ";

    private ZooKeeper zk;

    public void createConnection(String connectAddres, int sessionTimeOut)
    {
        try
        {
            zk = new ZooKeeper(connectAddres, sessionTimeOut, this);
            System.out.println(LOG_MAIN + "zk 开始启动连接服务器....");
            countDownLatch.await();
        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
    }

    public boolean createPath(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;
    }

    /**
     * 判断指定节点是否存在
     * 
     * @param path 节点路径
     */
    public Stat exists(String path, boolean needWatch)
    {
        try
        {
            return this.zk.exists(path, needWatch);
        }
        catch (Exception e)
        {
            e.printStackTrace();
            return null;
        }
    }

    public boolean updateNode(String path, String data) throws KeeperException, InterruptedException
    {
        exists(path, true);
        this.zk.setData(path, data.getBytes(), -1);
        return false;
    }

    public void process(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) throws KeeperException, InterruptedException
    {
        ZkClientWatcher zkClientWatcher = new ZkClientWatcher();
        zkClientWatcher.createConnection(CONNECT_ADDRES, SESSIONTIME);
//        boolean createResult = zkClientWatcher.createPath("/p15", "pa-644064");
        zkClientWatcher.updateNode("/pa2", "7894561");
    }

}

package com.hongmoshui.distributed;
import java.text.SimpleDateFormat;
import java.util.Date;

//生成订单类
public class OrderNumGenerator
{
    // 全局订单id
    public static int count = 0;

    public String getNumber()
    {
        try
        {
            Thread.sleep(200);
        }
        catch (Exception e)
        {
        }
        SimpleDateFormat simpt = new SimpleDateFormat("yyyy-MM-dd-HH-mm-ss");
        return simpt.format(new Date()) + "-" + ++count;
    }
}

package com.hongmoshui.distributed;

//使用多线程模拟生成订单号
public class OrderService implements Runnable
{
    private OrderNumGenerator orderNumGenerator = new OrderNumGenerator();

    public void run()
    {
        getNumber();
    }

    public void getNumber()
    {
        String number = orderNumGenerator.getNumber();
        System.out.println(Thread.currentThread().getName() + ",生成订单ID:" + number);
    }

    public static void main(String[] args)
    {
        System.out.println("####生成唯一订单号###");
        for (int i = 0; i < 100; i++)
        {
            new Thread(new OrderService()).start();
        }

    }
}

package com.hongmoshui.distributed;

//使用多线程模拟生成订单号
public class OrderSynchronizedService implements Runnable
{
    private OrderNumGenerator orderNumGenerator = new OrderNumGenerator();

    public void run()
    {
        getNumber();
    }

    public void getNumber()
    {
        synchronized (this)
        {
            String number = orderNumGenerator.getNumber();
            System.out.println(Thread.currentThread().getName() + ",生成订单ID:" + number);
        }
    }

    public static void main(String[] args)
    {
        System.out.println("####生成唯一订单号###");
        OrderService orderService = new OrderService();
        for (int i = 0; i < 100; i++)
        {
            new Thread(orderService).start();
        }

    }
}

package com.hongmoshui.distributed;

import java.util.concurrent.locks.ReentrantLock;

public class OrderLockService implements Runnable
{
    private OrderNumGenerator orderNumGenerator = new OrderNumGenerator();

    // 使用lock锁
    private java.util.concurrent.locks.Lock lock = new ReentrantLock();

    public void run()
    {
        getNumber();
    }

    public void getNumber()
    {
        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 void main(String[] args)
    {
        System.out.println("####生成唯一订单号###");
        OrderService orderService = new OrderService();
        for (int i = 0; i < 100; i++)
        {
            new Thread(orderService).start();
        }

    }
}

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

package com.hongmoshui.distributed;

public interface Lock
{
    // 获取到锁的资源
    public void getLock();

    // 释放锁
    public void unLock();
}

package com.hongmoshui.distributed;
import org.I0Itec.zkclient.ZkClient;

//将重复代码写入子类中..
public abstract class ZookeeperAbstractLock implements Lock
{
    // zk连接地址
    private static final String CONNECTSTRING = "127.0.0.1:2181";
    // 创建zk连接
    protected ZkClient zkClient = new ZkClient(CONNECTSTRING);
    protected static final String PATH = "/lock";

    public void getLock()
    {
        if (tryLock())
        {
            System.out.println("##获取lock锁的资源####");
        }
        else
        {
            // 等待
            waitLock();
            // 重新获取锁资源
            getLock();
        }
    }

    // 获取锁资源
    abstract boolean tryLock();
    // 等待
    abstract void waitLock();
    public void unLock()
    {
        if (zkClient != null)
        {
            zkClient.close();
            System.out.println("释放锁资源...");
        }
    }

}

package com.hongmoshui.distributed;

import java.util.concurrent.CountDownLatch;

import org.I0Itec.zkclient.IZkDataListener;

public class ZookeeperDistrbuteLock extends ZookeeperAbstractLock
{
    private CountDownLatch countDownLatch = null;

    @Override
    boolean tryLock()
    {
        try
        {
            zkClient.createEphemeral(PATH);
            return true;
        }
        catch (Exception e)
        {
//            e.printStackTrace();
            return false;
        }

    }

    @Override
    void waitLock()
    {
        IZkDataListener izkDataListener = new IZkDataListener()
        {

            public void handleDataDeleted(String path) throws Exception
            {
                // 唤醒被等待的线程
                if (countDownLatch != null)
                {
                    countDownLatch.countDown();
                }
            }

            public void handleDataChange(String path, Object data) throws Exception
            {

            }
        };
        // 注册事件
        zkClient.subscribeDataChanges(PATH, izkDataListener);
        if (zkClient.exists(PATH))
        {
            countDownLatch = new CountDownLatch(1);
            try
            {
                countDownLatch.await();
            }
            catch (Exception e)
            {
                e.printStackTrace();
            }
        }
        // 删除监听
        zkClient.unsubscribeDataChanges(PATH, izkDataListener);
    }

}

package com.hongmoshui.distributed;
import com.hongmoshui.OrderNumGenerator;

public class OrderService implements Runnable
{
    private OrderNumGenerator orderNumGenerator = new OrderNumGenerator();

    // 使用lock锁
    // private
    // java.util.concurrent.locks.Lock
    // lock = new ReentrantLock();
    private Lock lock = new ZookeeperDistrbuteLock();

    public void run()
    {
        getNumber();
    }

    public void getNumber()
    {
        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 void main(String[] args)
    {
        System.out.println("####生成唯一订单号###");
//        OrderService orderService = new OrderService();
        for (int i = 0; i < 100; i++)
        {
            new Thread(new OrderService()).start();
        }
    }
}

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

package com.hongmoshui.LoadBalance;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;

//ServerHandler
public class ServerHandler implements Runnable
{
    private Socket socket;

    public ServerHandler(Socket socket)
    {
        this.socket = socket;
    }

    public void run()
    {
        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;
            }
        }
    }
}

package com.hongmoshui.LoadBalance;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;

//##ServerScoekt服务端
public class ZkServerScoekt implements Runnable
{
    private int port = 18080;

    public static void main(String[] args) throws IOException
    {
        int port = 18080;
        ZkServerScoekt server = new ZkServerScoekt(port);
        Thread thread = new Thread(server);
        thread.start();
    }

    public ZkServerScoekt(int port)
    {
        this.port = port;
    }

    public void run()
    {
        ServerSocket serverSocket = null;
        try
        {
            serverSocket = new ServerSocket(port);
            System.out.println("Server start port:" + port);
            Socket socket = null;
            while (true)
            {
                socket = serverSocket.accept();
                new Thread(new ServerHandler(socket)).start();
            }
        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
        finally
        {
            try
            {
                if (serverSocket != null)
                {
                    serverSocket.close();
                }
            }
            catch (Exception e2)
            {

            }
        }
    }

}

package com.hongmoshui.LoadBalance;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;

public class ZkServerClient
{
    public static List<String> listServer = new ArrayList<String>();

    public static void main(String[] args)
    {
        initServer();
        ZkServerClient client = new ZkServerClient();
        BufferedReader console = new BufferedReader(new InputStreamReader(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 void initServer()
    {
        listServer.clear();
        listServer.add("127.0.0.1:18080");
    }

    // 获取当前server信息
    public static String getServer()
    {
        return listServer.get(0);
    }

    public void send(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(new InputStreamReader(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();
                }
            }
        }
    }
}

package com.hongmoshui.LoadBalance;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import org.I0Itec.zkclient.ZkClient;

public class ZkServerScoekt2 implements Runnable
{
    private static int port = 18081;

    public static void main(String[] args) throws IOException
    {
        ZkServerScoekt server = new ZkServerScoekt(port);
        Thread thread = new Thread(server);
        thread.start();
    }

    public ZkServerScoekt2(int port)
    {
        this.port = port;
    }

    public void regServer()
    {
        // 向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 void run()
    {
        ServerSocket serverSocket = null;
        try
        {
            serverSocket = new ServerSocket(port);
            regServer();
            System.out.println("Server start port:" + port);
            Socket socket = null;
            while (true)
            {
                socket = serverSocket.accept();
                new Thread(new ServerHandler(socket)).start();
            }
        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
        finally
        {
            try
            {
                if (serverSocket != null)
                {
                    serverSocket.close();
                }
            }
            catch (Exception e2)
            {

            }
        }
    }

}

package com.hongmoshui.LoadBalance;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;

public class ZkServerClient2
{
    public static List<String> listServer = new ArrayList<String>();

    public static void main(String[] args)
    {
        initServer();
        ZkServerClient client = new ZkServerClient();
        BufferedReader console = new BufferedReader(new InputStreamReader(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 void initServer()
    {
        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", new IZkChildListener()
        {

            public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception
            {
                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 static String getServer()
    {
        String serverName = listServer.get(count % serverCount);
        ++count;
        return serverName;
    }

    public void send(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(new InputStreamReader(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();
                }
            }
        }
    }
}