kafka学习安装与使用

请确保在启动Kafka服务器前，Zookeeper实例已经准备好并开始运行

1.Zookeeper搭建

1.1 Zookeeper

Zookeeper简单介绍: Zookeeper wiki
Zookeeper主页: Apache Zookeeper主页
IBM developerWorks: 分布式服务框架 Zookeeper -- 管理分布式环境中的数据

1.2 Zookeeper搭建方式

Zookeeper安装方式有三种，单机模式和集群模式以及伪集群模式。
单机模式：Zookeeper只运行在一台服务器上，适合测试环境；
伪集群模式：就是在一台物理机上运行多个Zookeeper实例；
集群模式：Zookeeper运行于一个集群上，适合生产环境，这个计算机集群被称为一个“集合体”（ensemble）

1.3 搭建步骤

采用伪集群的模式搭建.
1.下载Zookeeper: Zookeeper
2.解压Zookeeper文件
3.创建文件目录

image.png

4.在data文件下创建文本文件myid，里面只含数字，表明自己是哪台服务器，server0目录里的myid为0，server1目录里的myid为1，server2目录里的myid为2
5.分别修改三个zookeeper-3.4.6/conf/zoo_sample.cfg为zoo.cfg
6.修改zoo.cfg里的文件内容
server0/zookeeper-3.4.6/conf/zoo.cfg

# 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=/xxx/xxx/zk/server0/data
dataLogDir=/xxx/xxx/zk/server0/dataLog
# the port at which the clients will connect
clientPort=2181
# the maximum number of client connections.
# increase this if you need to handle more clients
#maxClientCnxns=60
#
# 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
#autopurge.purgeInterval=1

server.0=127.0.0.1:2880:3880
server.1=127.0.0.1:2881:3881
server.2=127.0.0.1:2882:3882

server1/zookeeper-3.4.6/conf/zoo.cfg

# 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=/xxx/xxxx/zk/server1/data
dataLogDir=/xxx/xxxx/zk/server1/dataLog
# the port at which the clients will connect
clientPort=2182
# the maximum number of client connections.
# increase this if you need to handle more clients
#maxClientCnxns=60
#
# 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
#autopurge.purgeInterval=1

server.0=127.0.0.1:2880:3880
server.1=127.0.0.1:2881:3881
server.2=127.0.0.1:2882:3882

server2/zookeeper-3.4.6/conf/zoo.cfg

# 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=/xxx/xxx/zk/server2/data
dataLogDir=/xxx/xxxx/zk/server2/dataLog
# the port at which the clients will connect
clientPort=2183
# the maximum number of client connections.
# increase this if you need to handle more clients
#maxClientCnxns=60
#
# 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
#autopurge.purgeInterval=1
server.0=127.0.0.1:2880:3880
server.1=127.0.0.1:2881:3881
server.2=127.0.0.1:2882:3882

每个配置相当于一台服务器，所以每个配置文件里的dataDir，dataLgDir, clientPort都不一样，每个配置文件的最后三行是一样的。
server.A=B：C：D:其中 A 是一个数字，就是myid里的那个数字，表示这个是第几号服务器；B 是这个服务器的 ip 地址，C和D是两个端口，C和D两个端口是用来交换信息与leader选举的。

Finally, note the two port numbers after each server name: “ 2888” and “3888”. Peers use the former port to connect to other peers. Such a connection is necessary so that peers can communicate, for example, to agree upon the order of updates. More specifically, a ZooKeeper server uses this port to connect followers to the leader. When a new leader arises, a follower opens a TCP connection to the leader using this port. Because the default leader election also uses TCP, we currently require another port for leader election. This is the second port in the server entry.

7.启动服务
进入每个目录/zookeeper-3.4.6，执行bin/zkServer.sh start，开启服务

image.png

8.接入客户端
可以进入server2/zookeeper-3.4.6，去连接server0，
执行bin/zkCli.sh -server 127.0.0.1:2181

image.png

2.kafka搭建

2.1 kafka 简介

kafka 主页
 IBM Apache kafka 工作原理介绍
 kafka入门介绍

2.2 搭建步骤

1.下载 kafka
2.解压 kafka
3.启动zk，可以用kafka自带的zookeeper(bin/zookeeper-server-start.sh config/zookeeper.properties &)
4.启动kafka server
bin/kafka-server-start.sh config/server.properties &
5.创建主题
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic 20170517
6.查看已经创建的主题
bin/kafka-topics.sh --list --zookeeper localhost:2181.
7.启动生产者
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic 20170517
8.启动消费者
另开一个终端，bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic 20170517 --from-beginning
在生产者下输入消息后回车，消费者就可以看到消息了
9.用监控工具查看kafka
下载 KafkaOffsetMonitor-assembly-0.2.0.jar
启动监控服务
java -cp KafkaOffsetMonitor-assembly-0.2.0.jar \
com.quantifind.kafka.offsetapp.OffsetGetterWeb \
--zk 127.0.0.1:2181 \
--port 8089 \
--refresh 10.seconds \
--retain 1.days

image.png