kafka 集群部署

#环境信息

server1 ip：172.17.0.2

server2 ip：172.17.0.3

server3 ip：172.17.0.4

 

#安装jre

3台机器都需要安装

下载地址：http://www.oracle.com/technetwork/java/javase/downloads/jre8-downloads-2133155.html

$ mkdir /usr/local/java

$ tar -xvf jre-8u161-linux-x64.tar -C /usr/local/java/

#设置环境变量

$ vim /etc/profile

export JAVA_HOME=/usr/local/java/jre1.8.0_161

export PATH=$PATH:$JAVA_HOME/bin

#配置生效

$ source /etc/profile

 

 

#安装集群

#为了测试方便，使用kafka自带的zookeeper

kafka下载地址：http://kafka.apache.org/downloads

 

#解压kafka 到 /usr/local/kafka中

 

#设置配置文件/etc/profile文件，增加kafka的路径

$ vim /etc/profile

export KAFKA_HOME=/usr/local/kafka

export PATH=$PATH:$KAFKA_HOME/bin

 

#配置文件立即生效

$ source /etc/profile

 

#修改配置

$ cd /usr/local/kafka/config/

$ vim server.properties

1、修改broker.id

broker.id=0     #集群中不能有重复的broker

2、修改listeners

listeners=PLAINTEXT://172.17.0.4:9092

 

log.dirs=/tmp/kafka-logs  #kafka数据的存放地址，多个地址的话用逗号分割，多个目录分布在不同磁盘上可以提高读写性能

log.retention.hours=168   #数据文件保留多长时间， 存储的最大时间超过这个时间会根据log.cleanup.policy设置数据清除策略

 

3、修改zookeeper.connect

zookeeper.connect=172.17.0.2:2181,172.17.0.3:2181,172.17.0.4:2181

 

#注意auto.create.topics.enable字段，若为true则如果producer写入某个不存在的topic时会自动创建该topic，若为false则需要事先创建否则会报错：failed after 3 retries。

 

#把server1的配置好的kafka拷贝到server2和server3上

$ scp -rp kafka root@172.17.0.3:/usr/local/

$ scp -rp kafka root@172.17.0.4:/usr/local/

 

#修改server2配置文件中的broker.id为1，server3主机的为2

 

#server1，server2和server3分别启动zookeeper 和 kafka

#启动zookeeper

$ bin/zookeeper-server-start.sh /usr/local/kafka/config/zookeeper.properties &

#停止zookeeper

$ bin/zookeeper-server-stop.sh

 

#启动kafka

$ bin/kafka-server-start.sh /usr/local/kafka/config/server.properties &

#停止kafka

$ bin/kafka-server-stop.sh

 

#测试集群

server1主机上创建一个名为test的topic

$ bin/kafka-topics.sh --create --zookeeper 172.17.0.2:2181 --replication-factor 1 --partitions 1 --topic test

 

server2和server3主机上创建consumer

$ bin/kafka-console-consumer.sh --zookeeper  172.17.0.3:2181 --topic test --from-beginning

$ bin/kafka-console-consumer.sh --zookeeper  172.17.0.4:2181 --topic test --from-beginning

 

#server1主机上创建producer

$ bin/kafka-console-producer.sh --broker-list 172.17.0.2:9092 --topic test

 

 

#server1主机上终端输入message，然后到server2和server3主机的终端查看

 

 

#删除topic

删除kafka存储目录（server.properties文件log.dirs配置，默认为"/tmp/kafka-logs"）相关topic目录

 

 

 

问题1：

WARN [Consumer clientId=consumer-1, groupId=console-consumer-950] Connection to node -1 could not be established. Broker may not be available.

解决办法：

这是因为配置文件中的PLAINTEXT跟你请求的内容不同。举例来说，在配置文件里配置的listeners=PLAINTEXT://172.17.0.2:9092，但是请求的是./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topic1 --from-beginning

正确的应该是./kafka-console-consumer.sh --bootstrap-server172.17.0.2:9092 --topic topic1 --from-beginning