我个人的kafka_2.12-1.1.0实践：安装与测试，单机版（★firecat推荐★）

一、我的服务器环境：CentOS7，kafka依赖JVM环境，要求安装有JDK

使用CentOS7系统，先删除自带的JDK，自带的不好用（例如jps终端命令不能用等）。下载官网完整的JDK，并设置环境变量JAVA_HOME。详情见：

https://blog.csdn.net/libaineu2004/article/details/80060812

[root@localhost ~]# java -version

openjdk version "1.8.0_141"

OpenJDK Runtime Environment (build 1.8.0_141-b16)

OpenJDK 64-Bit Server VM (build 25.141-b16, mixed mode)

二、下载

Kafka（http://kafka.apache.org/）依赖ZooKeeper

http://mirrors.shuosc.org/apache/zookeeper/zookeeper-3.4.10/zookeeper-3.4.10.tar.gz  --- 稳定版

http://mirrors.shuosc.org/apache/kafka/1.1.0/kafka-1.1.0-src.tgz

http://mirror.bit.edu.cn/apache/kafka/1.1.0/kafka_2.12-1.1.0.tgz

由于kafka_2.12-1.1.0.tgz是可执行程序，里面集成了zookeeper简易版，如果只是单机测试那么仅仅下载这个就可以了。但是简易版功能有限。在生产环境下，建议还是直接下载官方zookeeper软件。

三、运行与测试，以下都是终端命令

1、cd kafka_2.12-1.1.0

2、启动Zookeeper，如果无报错则说明启动成功。daemon以及nohup &是实现在后台启动，即守护进程。

方法1（推荐） ./bin/zookeeper-server-start.sh -daemon config/zookeeper.properties

方法2 nohup ./bin/zookeeper-server-start.sh config/zookeeper.properties&

3、启动Kafka，如果无报错则说明启动成功。daemon以及nohup &是实现在后台启动，即守护进程。

方法1 （推荐）./bin/kafka-server-start.sh -daemon config/server.properties

方法2 nohup ./bin/kafka-server-start.sh config/server.properties&

4、查看进程是否正常运行，通过检测2181与9092端口，2181是zookeeper，9092是kafka

netstat -tunlp|egrep "(2181|9092)"

tcp        0      0 :::2181                     :::*                        LISTEN      19787/java          

tcp        0      0 :::9092                     :::*                        LISTEN      28094/java

我们也可以使用jps终端命令查看是否启动：

[root@bogon kafka_2.12-1.1.0]# jps

4384 Kafka

3921 QuorumPeerMain

4412 Jps

5、创建主题，例如test

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

./bin/kafka-topics.sh --create --zookeeper 172.16.6.170:2181,172.16.6.170:2182,172.16.6.170:2183 --replication-factor 3 --partitions 3 --topic emqtest

--zookeeper : zookeeper集群列表，用英文逗号分隔。可以不用指定zookeeper整个集群内的节点列表，只指定某个或某几个zookeeper节点列表也是可以的

replication-factor ： 复制数目，提供failover机制；1代表只在一个broker上有数据记录，一般值都大于1，代表一份数据会自动同步到其他的多个broker，防止某个broker宕机后数据丢失。

partitions ： 一个topic可以被切分成多个partitions，一个消费者可以消费多个partitions，但一个partitions只能被一个消费者消费，所以增加partitions可以增加消费者的吞吐量。kafka只保证一个partitions内的消息是有序的，多个一个partitions之间的数据是无序的。

注意，

（1）使用集群时，必须手动先创建topic，指定zookeeper节点列表；单机也应该要创建。否则客户端会报错。

（2）主题名不建议使用"."和"_"字符。

[root@localhost kafka_2.12-1.0.0copy1]# ./bin/kafka-topics.sh --create --zookeeper 172.16.6.170:2181,172.16.6.170:2182,172.16.6.170:2183 --replication-factor 3 --

partitions 3 --topic emq_broker_message1

WARNING: Due to limitations in metric names, topics with a period ('.') or underscore ('_') could collide. To avoid issues it is best to use either, but not both.

Created topic "emq_broker_message1".

[root@localhost kafka_2.12-1.0.0copy1]# ./bin/kafka-topics.sh --create --zookeeper 172.16.6.170:2181,172.16.6.170:2182,172.16.6.170:2183 --replication-factor 3 --

partitions 3 --topic emqtest

Created topic "emqtest".

（3）建议创建主题时，指定分区数为kafka broker节点数目的整数倍，例如broker有3台，那么分区就设置为3个或6个或9个--partitions 3。创建完主题，可以在路径log.dirs看到主题和分区的文件夹。详情参见《Kafka入门与实践.牟大恩》3.6.1节描述

（4）--replication-factor 3，《kafka权威指南》6.3.1节建议复制系数为3.但是前提是数值不能大于broker的节点数。

如果broker节点数为1，复制系数设置为3，则会报错：

Error while executing topic command : Replication factor: 3 larger than available brokers: 1.

（5）单机时，zookeeper数目是1；集群时，zookeeper的数目建议为奇数个，至少3个。

（6）server.properties文件设置默认值

num.partitions=3

default.replication.factor=1

6、查看主题

./bin/kafka-topics.sh --describe --zookeeper  localhost:2181 --topic test

./bin/kafka-topics.sh --list --zookeeper localhost:2181

例如：broker集群，有3个节点。同时给主题mynewt分配了3个分区。每个分区都有leader和follow。

[root@bogon kafka_2.12-1.1.0]# ./bin/kafka-topics.sh --describe --zookeeper  localhost:2181 --topic mynewt

Topic:mynewt PartitionCount:3 ReplicationFactor:3 Configs:

Topic: mynewt Partition: 0 Leader: 2 Replicas: 2,1,0 Isr: 2,1,0

Topic: mynewt Partition: 1 Leader: 0 Replicas: 0,2,1 Isr: 0,2,1

Topic: mynewt Partition: 2 Leader: 1 Replicas: 1,0,2 Isr: 1,0,2

7、开启一个终端，发送消息，生产者的消息要发往kafka

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

8、另起一个终端，消费消息，消费者的消息来自zookeeper（协调转发）

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

将来新版本需要写成./bin/kafka-console-consumer.sh --bootstrap-server localhost:2181 --topic test --from-beginning

9、结束进程

./bin/kafka-server-stop.sh

./bin/zookeeper-server-stop.sh

kafka启动时先启动zookeeper，再启动kafka；关闭时相反，先关闭kafka，再关闭zookeeper。

四、总结

1、kafka有什么？

producer 消息的生成者，即发布消息

consumer 消息的消费者，即订阅消息

broker Kafka以集群的方式运行，可以由一个或多个服务组成，服务即broker

zookeeper 协调转发

2、配置文件与消息持久化路径

/config/server.properties

# A comma seperated list of directories under which to store log files

log.dirs=/tmp/kafka-logs

例如客户端发送test为主题的消息，会持久化在这个文件：/tmp/kafka-logs/test-0/00000000000000000000.log

主题列表会记录在/tmp/zookeeper

/config/zookeeper.properties

# the directory where the snapshot is stored.

dataDir=/tmp/zookeeper

# the port at which the clients will connect

clientPort=2181

建议修改kafka的日志目录和zookeeper数据目录，因为这两项默认放在tmp目录，而tmp目录中内容会随重启而丢失：

server.properties：

   log.dirs=/tmp/kafka-logs

   修改为

   log.dirs=/usr/local/kafka/logs

zookeeper.properties

   dataDir=/tmp/zookeeper

   修改为

   dataDir=/usr/local/zookeeper/data

3、配置文件bootstrap.servers 和 broker.list区别：

本以为是两个参数，其实是实现一个功能，查看源代码后发现broker.list是旧版本命令

4、自己写的shell启动和关闭kafka脚本，kafka.sh -- 这个脚本有问题，不稳定

#!/bin/bash
kafka_home=/root/Downloads/kafka_2.12-1.0.0/
case $1 in 
   start)  # 服务启动需要做的步骤
           echo "zookeeper start"
           $kafka_home/bin/zookeeper-server-start.sh -daemon $kafka_home/config/zookeeper.properties
           #sleep 1
           echo "kafka start"
           $kafka_home/bin/kafka-server-start.sh -daemon $kafka_home/config/server.properties
           #sleep 1
           ;;
   stop)   # 服务停止需要做的步骤
           echo "kafka stop"
           $kafka_home/bin/kafka-server-stop.sh
           #sleep 1
           echo "zookeeper stop"
           $kafka_home/bin/zookeeper-server-stop.sh
           #sleep 1
           ;;
   restart) # 重启服务需要做的步骤
            ...
           ;;
   status) # 查看状态需要做的步骤
             ...
           ;;
   *) echo "$0 {start|stop|restart|status}"
           exit 4
           ;;
esac

5、zookeeper集群

kafka是通过zookeeper来管理集群。 kafka软件包内虽然包括了一个简版的zookeeper，但是感觉功能有限。在生产环境下，建议还是直接下载官方zookeeper软件。 下载最新版的zookeeper软件：

http://zookeeper.apache.org/

http://mirror.bit.edu.cn/apache/zookeeper/

zookeeper-3.4.11.tar.gz

欢迎访问姊妹篇：我个人的kafka broker和zookeeper集群实践（★firecat推荐★）

---

参考文献

物联网架构成长之路(8)-EMQ-Hook了解、连接Kafka发送消息

centos7 安装 kafka_2.11-1.0.0

zookeeper安装及部署

Kafka安装及部署

CentOS下ZooKeeper单机模式、集群模式安装

Kafka单机、集群模式安装详解(一)

kafka集群监控之KafkaOffsetMonitor 0.4.1版本 以及用sbt编译KafkaOffsetMonitor

https://github.com/quantifind/KafkaOffsetMonitor