分布式环境ZooKeeper3.4.5安装和JavaAPI

 

 

VirtualBox+Centos6.4搭建Hadoop1.1.2分布式环境http://mvplee.iteye.com/blog/2233435

 

确保集群内服务器时间一致

 

解压文件

[root@hadoop1 local]# tar -zxvf zookeeper-3.4.5.tar.gz 

配置home变量

[root@hadoop1 local]# more /etc/profile
export ZOOKEEPER_HOME=/usr/local/zookeer-3.4.5
PATH=.:$HADOOP_HOME/bin:$ZOOKEEPER_HOME/bin:$JAVA_HOME/bin:$PATH
[root@hadoop1 local]# source /etc/profile

 配置zookeeper配置文件，复制一副sample配置文件，对其进行修改

[root@hadoop1 conf]# pwd
/usr/local/zookeeper-3.4.5/conf
[root@hadoop1 conf]# mv zoo_sample.cfg zoo.cfg

文件内可以配置的参数有

1.tickTime：CS通信心跳时间
Zookeeper 服务器之间或客户端与服务器之间维持心跳的时间间隔，也就是每个 tickTime 时间就会发送一个心跳。tickTime以毫秒为单位。
tickTime=2000  

2.initLimit：LF初始通信时限
集群中的follower服务器(F)与leader服务器(L)之间初始连接时能容忍的最多心跳数（tickTime的数量）。
initLimit=5  

3.syncLimit：LF同步通信时限
集群中的follower服务器与leader服务器之间请求和应答之间能容忍的最多心跳数（tickTime的数量）。
syncLimit=2  
 
4.dataDir：数据文件目录
Zookeeper保存数据的目录，默认情况下，Zookeeper将写数据的日志文件也保存在这个目录里。
dataDir=/home/michael/opt/zookeeper/data  

5.clientPort：客户端连接端口
客户端连接 Zookeeper 服务器的端口，Zookeeper 会监听这个端口，接受客户端的访问请求。
clientPort=2181 

6.服务器名称与地址：集群信息（服务器编号，服务器地址，LF通信端口，选举端口）
这个配置项的书写格式比较特殊，规则如下：
server.N=YYY:A:B 

 修改内容

[root@hadoop1 conf]# more zoo.cfg
dataDir=/usr/local/zookeeper-3.4.5/data
server.1=hadoop1:2888:3888
server.2=hadoop2:2888:3888
server.3=hadoop3:2888:3888

 配置zookeeper节点

[root@hadoop1 zookeeper-3.4.5]# mkdir data
[root@hadoop1 data]# vim myid
[root@hadoop1 data]# more myid
0

 把hadoop1上所配置的复制到hadoop2、hadoop3上

[root@hadoop1 zookeeper-3.4.5]# scp /etc/profile hadoop2:/etc/profile
[root@hadoop1 zookeeper-3.4.5]# scp /etc/profile hadoop3:/etc/profile
[root@hadoop1 zookeeper-3.4.5]# scp -r /usr/local/zookeeper-3.4.5 hadoop2:/usr/local/
[root@hadoop1 zookeeper-3.4.5]# scp -r /usr/local/zookeeper-3.4.5 hadoop3:/usr/local/

 在hadoop2、hadoop3上修改zookeeper-3.4.5/conf/data/myid节点配置文件

 

在hadoop1上启动zookeeper

[root@hadoop1 bin]# pwd
/usr/local/zookeeper-3.4.5/bin
[root@hadoop1 bin]# zkServer.sh start
JMX enabled by default
Using config: /usr/local/zookeeper-3.4.5/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED

 启动后会在目录下多出一个zookeeper.out文件，记录异常信息java.net.ConnectException:

[root@hadoop1 bin]# more zookeeper.out
2015-08-07 09:13:45,063 [myid:1] - WARN  [QuorumPeer[myid=1]/0:0:0:0:0:0:0:0:2181:QuorumCnxManager@368] - Cannot open channel to 2 at election address hadoop
2/192.168.56.22:3888
java.net.ConnectException: Connection refused
2015-08-07 09:14:36,271 [myid:1] - WARN  [QuorumPeer[myid=1]/0:0:0:0:0:0:0:0:2181:QuorumCnxManager@368] - Cannot open channel to 3 at election address hadoop
3/192.168.56.23:3888
java.net.ConnectException: Connection refused

 原因是hadoop2、hadoop3还没启动，所以通信失败，启动hadoop2、hadoop3上的zookeeper

[root@hadoop2 bin]# zkServer.sh start
[root@hadoop3 bin]# zkServer.sh start

 zookeeper都启动会搞选举，查看选举结果

[root@hadoop1 bin]# zkServer.sh status
JMX enabled by default
Using config: /usr/local/zookeeper-3.4.5/bin/../conf/zoo.cfg
Mode: follower
[root@hadoop2 bin]# zkServer.sh status
JMX enabled by default
Using config: /usr/local/zookeeper-3.4.5/bin/../conf/zoo.cfg
Mode: leader
[root@hadoop3 bin]# zkServer.sh status
JMX enabled by default
Using config: /usr/local/zookeeper-3.4.5/bin/../conf/zoo.cfg
Mode: follower

 

Zookeeper的JavaAPI操作

package example.zookeeper;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.List;
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.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

public class ZKClientTest {
	private static final String HOSTS = "hdp01:2181,hdp02:2181,hdp03:2181";
	private static final int TIMEOUT = 2000;
	// latch.await()方法执行时，方法所在的线程会等待，当latch的count减为0时，才会唤醒等待的线程
	private static final CountDownLatch LATCH = new CountDownLatch(1);
	private static ZooKeeper zkClient = null;

	@Before
	public void init() throws IOException, InterruptedException {
		zkClient = new ZooKeeper(HOSTS, TIMEOUT, new Watcher() {
			// 事件监听回调方法
			public void process(WatchedEvent event) {
				if (LATCH.getCount() > 0 && event.getState() == KeeperState.SyncConnected) {
					System.out.println("KeeperState is SyncConnected");
					LATCH.countDown();
				}
				System.out.println("getType:" + event.getType());
				System.out.println("getPath:" + event.getPath());
				System.out.println("getState:" + event.getState());
				// zkClient.getData("", true, null);//持续监听
			}
		});
		LATCH.await();
	}

	@After
	public void destroy() throws InterruptedException {
		zkClient.close();
		System.out.println("zkClient is closed");
	}

	@Test
	public void testCreate() throws KeeperException, InterruptedException {
		// 参数1：要创建的节点的路径
		// 参数2：节点的数据
		// 参数3：节点的权限
		// 参数4：节点的类型
		String create = zkClient.create("/eclipse", "helloZK".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
		System.out.println(create);
	}

	@Test
	public void testCreateChildren() throws KeeperException, InterruptedException {
		zkClient.create("/eclipse/ide1", "ide1".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
		zkClient.create("/eclipse/ide2", "ide2".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
	}

	@Test
	public void testGetChildren() throws KeeperException, InterruptedException {
		List<String> children = zkClient.getChildren("/eclipse", true);
		for (String string : children) {
			System.out.println(string);
		}
	}

	@Test
	public void testGetData() throws KeeperException, InterruptedException, UnsupportedEncodingException {
		// watch==true为使用new ZooKeeper()构造方法中的Watcher
		byte[] bytes = zkClient.getData("/eclipse", true, null);
		System.out.println(new String(bytes, "UTF-8"));
	}

	@Test
	public void testDelete() throws InterruptedException, KeeperException {
		// -1 删除任何节点的版本数据
		zkClient.delete("/eclipse/ide", -1);
	}

	@Test
	public void testSetData() throws KeeperException, InterruptedException {
		// -1 任何节点的版本数据
		zkClient.setData("/eclipse/ide1", "i'm a ide".getBytes(), -1);
	}

	@Test
	public void testExist() throws KeeperException, InterruptedException {
		Stat stat = zkClient.exists("/none", true);
		// 节点不存在Stat对象为null
		System.out.println(stat);
	}
}