Apache Kafka是一种颇受欢迎的分布式消息代理系统，旨在有效地处理大量的实时数据。Kafka集群不仅具有高度可扩展性和容错性，而且与其他消息代理（如ActiveMQ和RabbitMQ）相比，还具有更高的吞吐量。虽然它通常用作pub/sub消息传递系统，但许多组织也将其用于日志聚合，因为它为发布的消息提供持久存储。

您可以在一台服务器上部署Kafka，也可以构建一个分布式的Kafka集群来提高性能。本文介绍如何在多节点CentOS 7服务器实例上安装Apache Kafka。

先决条件：

欲安装kafka集群服务器，首先要安装以下组件：

《Linux JAVA JDK JRE 环境变量安装与配置》
《在 Linux 多节点安装配置 Apache Zookeeper 分布式集群》

服务器列表：

10.10.204.63
10.10.204.64
10.10.204.65

1.安装

创建用户和组：

 # groupadd kafka
 # useradd -g kafka -s /sbin/nologin kafka

下载Kafka包：

 # cd /usr/local
 # wget //apache.fayea.com/kafka/0.10.2.1/kafka_2.10-0.10.2.1.tgz

解压创建软连接：

 # tar zxvf kafka_2.10-0.10.2.1.tgz
 # ln -s kafka_2.10-0.10.2.1 kafka

设置权限及创建Kafka日志存放目录：

 # chown -R kafka:kafka kafka_2.10-0.10.2.1 kafka
 # mkdir -p /usr/local/kafka/logs

添加系统变量：

编辑：/etc/profile 文件，在最下面添加以下内容：

 export KAFKA_HOME=/usr/local/kafka_2.10-0.10.2.1
 export PATH=$KAFKA_HOME/bin:$PATH

使变量生效：

 # source /etc/profile

2.配置

修改添加Kafka服务器的配置文件：

 # cd /usr/local/kafka/config
 # vim server.properties

#唯一值，每个server填写不一样。
broker.id=63
#允许删除主题。
delete.topic.enable=true
#修改；协议、当前broker机器ip、端口，此值可以配置多个，跟SSL等有关系。
listeners=PLAINTEXT://10.10.204.63:9092
#修改；kafka数据的存放地址，多个地址的话用逗号分割,例如 /data/kafka-logs-1,/data/kafka-logs-2。
log.dirs=/usr/local/kafka/logs/kafka-logs
#每个topic的分区个数，若是在topic创建时候没有指定的话会被topic创建时的指定参数覆盖。
num.partitions=3
#新增；表示消息体的最大大小，单位是字节。
message.max.bytes=5242880
#新增；是否允许自动创建topic，若是false，就需要通过命令创建topic。
default.replication.factor=2
#新增；replicas每次获取数据的最大大小。
replica.fetch.max.bytes=5242880
#新增；配置文件中必须使用以下配置，否则只会标记为删除，而不是真正删除。
delete.topic.enable=true
#新增；是否允许 leader 进行自动平衡，boolean 值，默认为 true。
auto.leader.rebalance.enable=true
#kafka连接的zk地址，各个broker配置一致。
zookeeper.connect=10.10.204.63:2181,10.10.204.64:2181,10.10.204.65:2181

#可选配置
#是否允许自动创建 topic，boolean 值，默认为 true。
auto.create.topics.enable=true
#指定 topic 的压缩方式，string 值，可选有。
compression.type=high
#会把所有的日志同步到磁盘上，避免重启之后的日志恢复，减少重启时间。
controlled.shutdown.enable=true

注：broker的配置文件中有zookeeper的地址,也有自己的broker ID, 当broker启动后,会在zookeeper中新建一个znode。

修改其他配置文件：

 # vim zookeeper.properties

修改为：

 dataDir=/usr/local/zookeeper/data

新增：

 server.1=10.10.204.63:2888:3888
 server.2=10.10.204.64:2888:3888
 server.3=10.10.204.65:2888:3888

修改以下配置文件：

# vim producer.properties

bootstrap.servers=10.10.204.63:9092,10.10.204.64:9092,10.10.204.65:9092

# vim consumer.properties

zookeeper.connect=10.10.204.63:2181,10.10.204.64:2181,10.10.204.65:2181

3.启动

启动所有节点kafka服务（可以通过查看日志，或者检查进程状态，保证Kafka集群启动成功）：

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

执行上述命令后，会出来滚动的启动信息，直至窗口静止，此时需重开终端检查是否启动成功；

# jps
 9939 Jps
 2201 QuorumPeerMain
 2303 Kafka

4.使用测试

下面操作可以在任意节点，重新打开一个终端操作：

执行以下命令，建立一个名为 renwole 的topic。

 # cd /usr/local/kafka/bin
 # ./kafka-topics.sh --create --zookeeper 10.10.204.63:2181,10.10.204.64:2181,10.10.204.65:2181 --replication-factor 1 --partitions 1 --topic renwole
 Created topic "renwole".

解释:

 --replication-factor 1 复制1份
 --partitions 1 创建1个分区
 --topic 主题为renwole

查看已创建的topic：

# ./kafka-topics.sh --list --zookeeper 10.10.204.63:2181
 _consumer_offsets
 renwole

注：可以配置 broker 自动创建 topic。

发送消息（Kafka 使用一个简单的命令行producer（然后可以随意输入内容，回车可以发送，ctrl+c 退出）默认的每条命令将发送一条消息。）：

# ./kafka-console-producer.sh --broker-list 10.10.204.64:9092 --topic renwole

在消息接收端，执行以下命令查看收到的消息：

# ./kafka-console-consumer.sh --bootstrap-server 10.10.204.63:9092 --topic renwole --from-beginning

执行以下命令删除topic：

# ./kafka-topics.sh --delete --zookeeper 10.10.204.63:2181,10.10.204.64:2181,10.10.204.65:2181 --topic renwole

5.查看集群状态

kafka已经成功完成安装，查看kafka集群节点ID状态：

注：可在任意节点连接zookeeper客户端。

 # cd /usr/local/zookeeper/bin
 # ./zkCli.sh
 Connecting to localhost:2181
 [myid:] - INFO [main:Environment@100] - Client environment:zookeeper.version=3.4.10-39d3a4f269333c922ed3db283be479f9deacaa0f, built on 03/23/2017 10:13 GMT
 [myid:] - INFO [main:Environment@100] - Client environment:host.name=10-10-204-63.10.10.204.63
 [myid:] - INFO [main:Environment@100] - Client environment:java.version=1.8.0_144
 [myid:] - INFO [main:Environment@100] - Client environment:java.vendor=Oracle Corporation
 [myid:] - INFO [main:Environment@100] - Client environment:java.home=/usr/java/jdk1.8.0_144/jre
 [myid:] - INFO [main:Environment@100] - Client environment:java.class.path=/usr/local/zookeeper/bin/../build/classes:/usr/local/zookeeper/bin/../build/lib/*.jar:/usr/local/zookeeper/bin/../lib/slf4j-log4j12-1.6.1.jar:/usr/local/zookeeper/bin/../lib/slf4j-api-1.6.1.jar:/usr/local/zookeeper/bin/../lib/netty-3.10.5.Final.jar:/usr/local/zookeeper/bin/../lib/log4j-1.2.16.jar:/usr/local/zookeeper/bin/../lib/jline-0.9.94.jar:/usr/local/zookeeper/bin/../zookeeper-3.4.10.jar:/usr/local/zookeeper/bin/../src/java/lib/*.jar:/usr/local/zookeeper/bin/../conf:.:/usr/java/jdk1.8.0_144/jre/lib/rt.jar:/usr/java/jdk1.8.0_144/lib/dt.jar:/usr/java/jdk1.8.0_144/lib/tools.jar:/usr/java/jdk1.8.0_144/jre/lib
 [myid:] - INFO [main:Environment@100] - Client environment:java.library.path=/usr/java/packages/lib/amd64:/usr/lib64:/lib64:/lib:/usr/lib
 [myid:] - INFO [main:Environment@100] - Client environment:java.io.tmpdir=/tmp
 [myid:] - INFO [main:Environment@100] - Client environment:java.compiler=
 [myid:] - INFO [main:Environment@100] - Client environment:os.name=Linux
 [myid:] - INFO [main:Environment@100] - Client environment:os.arch=amd64
 [myid:] - INFO [main:Environment@100] - Client environment:os.version=3.10.0-514.21.2.el7.x86_64
 [myid:] - INFO [main:Environment@100] - Client environment:user.name=root
 [myid:] - INFO [main:Environment@100] - Client environment:user.home=/root
 [myid:] - INFO [main:Environment@100] - Client environment:user.dir=/usr/local/zookeeper-3.4.10/bin
 [myid:] - INFO [main:ZooKeeper@438] - Initiating client connection, connectString=localhost:2181 sessionTimeout=30000 watcher=org.apache.zookeeper.ZooKeeperMain$MyWatcher@69d0a921
 Welcome to ZooKeeper!
 [myid:] - INFO [main-SendThread(localhost:2181):ClientCnxn$SendThread@1032] - Opening socket connection to server localhost/0:0:0:0:0:0:0:1:2181. Will not attempt to authenticate using SASL (unknown error)
 JLine support is enabled
 [myid:] - INFO [main-SendThread(localhost:2181):ClientCnxn$SendThread@876] - Socket connection established to localhost/0:0:0:0:0:0:0:1:2181, initiating session
 [myid:] - INFO [main-SendThread(localhost:2181):ClientCnxn$SendThread@1299] - Session establishment complete on server localhost/0:0:0:0:0:0:0:1:2181, sessionid = 0x35ddf80430b0008, negotiated timeout = 30000
 WATCHER::
 WatchedEvent state:SyncConnected type:None path:null
 [zk: localhost:2181(CONNECTED) 0] ls /brokers/ids #查看ID
 [63, 64, 65]

可以看到三台kafka实例ID都在线，如果模拟任意一台节点掉线，查看结果就会不同（这个ID号是前面设置的 broker.id ）。

开放端口加入防火墙：

# firewall-cmd --permanent --add-port=9092/tcp
# firewall-cmd --reload

6.开机启动

创建system单元文件：

在 /usr/lib/systemd/system 目录下创建 kafka.service 填写以下内容：

[Unit]
 Description=Apache Kafka server (broker)
 Documentation=//kafka.apache.org/documentation/
 Requires=network.target remote-fs.target
 After=network.target remote-fs.target

[Service]
 Type=simple
 Environment="LOG_DIR=/usr/local/kafka/logs"
 User=kafka
 Group=kafka
 #Environment=JAVA_HOME=/usr/java/jdk1.8.0_144
 ExecStart=/usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.properties
 ExecStop=/usr/local/kafka/bin/kafka-server-stop.sh
 Restart=on-failure
 SyslogIdentifier=kafka

[Install]
 WantedBy=multi-user.target

启动kafka实例服务器：

# systemctl daemon-reload
# systemctl enable kafka.service
# systemctl start kafka.service
# systemctl status kafka.service

7.此外Kafka也可以通过 Web UI 界面进行集群的管理，你可以参阅：

《安装配置 Kafka Manager 分布式管理工具》

滴滴，好了，到目前为止，kafka分布式消息队列已经部署完成，其中包括必要的优化信息。目前，您可以用于大多数编程语言的Kafka客户端去创建Kafka生产者和消费者，轻松地将其用于您的项目中。

参考资料：
//kafka.apache.org/documentation.html#quickstart
//www.ibm.com/developerworks/cn/opensource/os-cn-kafka/
//tech.meituan.com/kafka-fs-design-theory.html
//blog.jobbole.com/99195/