一个典型的Kafka集群中包含若干Producer（可以是Web前端FET，或者是服务器日志等），若干Broker（Kafka支持水平扩展，一般Broker数量越多，集群吞吐率越高），若干ConsumerGroup，以及一个ZooKeeper集群。Kafka通过Zookeeper管理Kafka集群配置：选举Kafka Broker的Leader，以及在Consumer Group发生变化时进行Rebalance，因为Consumer消费Kafka topic的Partition的offsite信息是存在ZooKeeper的。Producer使用push模式将消息发布到Broker，Consumer使用pull模式从Broker订阅并消费消息。

在node1节点上安装配置Kafka，然后再使用【scp】命令，将安装目录分发到其他节点（node2、node3）

1、可以从官网下载Kafka安装包 ，当前环境已经提供了Kafka的安装包，存放在 /opt/software目录下。

• 在node1上解压安装Kafka

[root@node1 ~]# cd /opt/software
[root@node1 software]# tar -xzf kafka.tar.gz -C /opt/module/

2、在node1系统环境变量/etc/profile

[root@node1 software]# vi /etc/profile

• 在文件末尾添加如下配置：

export KAFKA_HOME=/opt/module/kafka
export PATH=$PATH:$KAFKA_HOME/bin

• 使用【source】命令，使/etc/profile配置生效

[root@node1 software]# source /etc/profile

3、配置Kafka

• 为Kafka创建数据目录

[root@node1 software]# cd $KAFKA_HOME
[root@node1 kafka]# mkdir kafka-logs

• 然后打开config目录下的server.properties文件，修改日志目录为刚刚创建的目录:

[root@node1 kafka]# cd $KAFKA_HOME/config
[root@node1 config]# vi server.properties

• 找到并修改如下配置的值，其中broker.id的值需要保证在整个集群中是唯一的

broker.id=1
log.dirs=/opt/module/kafka/kafka-logs
zookeeper.connect=node1:2181,node2:2181,node3:2181

• 将Kafka和环境变量拷贝到node2、node3并修改broker.id的值

解释

[root@node1 config]# cd /opt/module/ [root@node1 module]# scp -rq kafka node2:/opt/module/ [root@node1 module]# scp -rq kafka node3:/opt/module/ [root@node1 module]# scp -rq /etc/profile node2:/etc/ [root@node1 module]# scp -rq /etc/profile node3:/etc/

• 在node2修改kafka的配置文件server.properties，将broker.id的值修改为2

[root@node2 ~]# source /etc/profile
[root@node2 ~]# cd $KAFKA_HOME/config 
[root@node2 config]# vi server.properties

• 修改后broker.id的值如下

• 在node3修改kafka的配置文件server.properties，将broker.id的值修改为3

[root@node3 ~]# source /etc/profile
[root@node3 ~]# cd $KAFKA_HOME/config 
[root@node3 config]# vi server.properties

• 修改后broker.id的值如下

• 分别在node1、node2、node3启动Kafka的后台守护进程

[root@node1 ~]# kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties

[root@node2 ~]# kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties

[root@node3 ~]# kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties

4、验证Kafka

• 创建topic，创建名为test的topic，分区数1，副本1

[root@node1 ~]# kafka-topics.sh --create --zookeeper node1:2181,node2:2181,node3:2181 --replication-factor 1 --partitions 1 --topic test

• 查看topic的状态

[root@node1 ~]# kafka-topics.sh --describe --zookeeper node1:2181,node2:2181,node3:2181 --topic test

• 在client客户端节点上打开一个终端，使用ssh命令连接到node1，发送消息

[root@client~]# ssh node1
[root@node1 ~]# kafka-console-producer.sh --broker-list node1:9092 --topic test

• 在client客户端节点上重新打开一个终端，使用ssh命令连接到node1的消费消息

[root@client~]# ssh node1
[root@node1 ~]# kafka-console-consumer.sh --bootstrap-server node1:9092 --from-beginning --topic test