kafka1.0+ 集群搭建

由于kafka已经下降了zookeeper在kafka中的重要性,全部不推荐单独搭建zookeeper了,使用集成的zookeeper已经很是好了

 

1.官网下载kafka最新版本

如今kafka版本是1.1.0,可是根据scala编译分为2.11和2.12版本,其实就是2.11对应JDK1.7,   2.12对应JDK1.8

 

2.配置

例如3台集群主机IP分别为10.128.90.111;10.128.90.112;10.128.90.113

kafka解压部署在主机路径 /app/kafka_2.11-1.1.0

(1).建立kafka_data文件夹: /app/kafkadata(文件夹名字本身取)

(2).建立zookeeper_data文件夹:/app/zookeeperdata(文件夹名字本身取)

(3).修改kafka_2.11-1.1.0/conf文件夹下server.properties文件

broker.id=0   //kafka broker的id，三个节点不相同

listeners=PLAINTEXT://10.128.90.111:29092    //kafka端口,三个节点分别填写节点本机IP

log.dir=/app/kafkadata       //kafka日志路径

zookeeper.connect=10.128.90.111:22181,10.128.90.112:22181,10.128.90.113:22181  //zookeeper集群地址，使用,分隔

(4).修改kafka_2.11-1.1.0/conf文件夹下zookeeper.properties文件

dataDir=/app/zookeeperdata        //zookeeper快照存储地址

clientPort=22181                           //端口号

maxClientCnxns=0                        //客户端与服务端链接限制

initLimit=10

syncLimit=5

server.0=10.128.90.111:22888:23888
server.1=10.128.90.112:22888:23888
server.2=10.128.90.113:22888:23888

22888用于follower与leader之间的数据同步与通讯；23888用于leader选举时的通讯

0,1,2  是zookeeper三个节点id,需在zookeeper快照存储文件夹中建立myid文件,并写入对应的id号

例如 10.128.90.111 需写入0

 

(5).建立myid文件

切换路径到/app/zookeeperdata下

vim myid

写入0保存

 

3.启动

启动zookeeper：
        bin/zookeeper-server-start.sh   -daemon   config/zookeeper.properties &

启动kafka：
         bin/kafka-server-start.sh   -daemon  config/server.properties &

 

4.Topic

建立Topic

bin/kafka-topics.sh --create --topic mytest --zookeeper 10.128.90.111:22181,10.128.90.112:22181,10.128.90.113:22181 --replication-factor 1 --partitions 1

replication-factor 副本数,至关于把数据复制了几份保存在服务器上

partitions 分区,至关于把数据打散保存,能够提升数据的吞吐量,不易过大,过大会形成稳定性,须要更大的内存,需打开更多的句柄

 

查看Topic列表

 bin/kafka-topics.sh --list --zookeeper 10.128.90.111:22181

 

5.测试

建立生产者

bin/kafka-console-producer.sh --broker-list 10.128.90.111:29092,10.128.90.112:29092,10.128.90.113:29092 --topic mytest

建立消费者

bin/kafka-console-consumer.sh --bootstrap-server 10.128.90.111:29092,10.128.90.112:29092,10.128.90.113:29092 --topic mytest

这时在生产者发生消息,消费者窗口就能接收到消息了

 

6.指定JDK

例如

修改bin/kafka-server-start.sh

第一行加 export JAVA_HOME=/app/jdk1.8.0_131