1 kafka开发环境

1.1 安装Java环境

1.1.1 下载linux下的安装包

登陆网址https://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html

下载完成后，Linux默认下载位置在当前目录下的Download或下载文件夹下，通过命令cd ~/Downloads或cd ~/下载即可查看到对应的文件。

1.1.2 解压安装包jdk-8u202-linux-x64.tar.gz

tar -zxvf jdk-8u291-linux-x64.tar.gz

解压后的文件夹为jdk1.8.0_291 进入文件夹和查看文件

cd jdk1.8.0_291

1.1.3 将解压后的文件移到/usr/lib目录下

在/usr/bin目录下新建jdk目录

sudo mkdir /usr/lib/jdk

将解压的jdk文件移动到新建的/usr/lib/jdk目录下来

sudo mv jdk1.8.0_291 /usr/lib/jdk/

1.1.4 配置java环境变量

这里是将环境变量配置在etc/profile，即为所有用户配置JDK环境。 使用命令打开/etc/profile文件

sudo vim /etc/profile

在末尾添加以下几行文字：

#set java env
export JAVA_HOME=/usr/lib/jdk/jdk1.8.0_291
export JRE_HOME=${JAVA_HOME}/jre
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib
export PATH=${JAVA_HOME}/bin:$PATH

1.1.5 执行命令使修改立即生效

source /etc/profile

1.1.6 测试安装是否成功

在终端输入，出现版本号说明安装成功。

java -version

1.2 Kafka的安装部署

1.2.1 下载kafka

wget https://archive.apache.org/dist/kafka/2.0.0/kafka_2.11-2.0.0.tgz

1.2.2 安装kafka

我们下载的kafka是已经编译好的程序，只需要解压即可得到执行程序。

tar -zxvf kafka_2.11-2.0.0.tgz

进入kafka目录，以及查看对应的文件和目录

cd kafka_2.11-2.0.0

ls

 bin：为执行程序         config：为配置文件         libs：为库文件

1.2.3 配置和启动zookeeper

下载的kafka程序里自带了zookeeper，kafka自带的Zookeeper程序脚本与配置文件名与原生 Zookeeper稍有不同。 kafka自带的Zookeeper程序使用bin/zookeeper-server-start.sh，以及bin/zookeeper-server-stop.sh 来启动和停止Zookeeper。 kafka依赖于zookeeper来做master选举一起其他数据的维护。

启动zookeeper：zookeeper-server-start.sh

停止zookeeper：zookeeper-server-stop.sh

所以我们可以通过下面的脚本来启动zk服务，当然，也可以自己独立搭建zk的集群来实现。这里我们直接使用kafka自带的zookeeper。

启动zookeeper

前台运行：sh zookeeper-server-start.sh ../config/zookeeper.properties

后台运行：sh zookeeper-server-start.sh -daemon ../config/zookeeper.properties

默认端口为：2181，可以通过命令lsof -i:2181 查看zookeeper是否启动成功。

 1.2.4 启动和停止kafka

修改server.properties（在config目录）, 增加zookeeper的配置，这里只是本地的配置，如果是另一台机器运行zookeeper，要配置对应的ip地址。

zookeeper.connect=localhost:2181

启动kafka

sh kafka-server-start.sh -daemon ../config/server.properties

默认端口为：9092，可以通过命令lsof -i:9092查看kafka是否启动成功。

停止kafka

sh kafka-server-stop.sh -daemon ../config/server.properties

1.3 kafka的基本操作

1.3.1 创建topic

sh kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1

--topic test

成功则显示：

Created topic "test".

参数说明：

--create 是创建主题的的动作指令

--zookeeper 指定kafka所连接的zookeeper服务地址

--replicator-factor 指定了副本因子（即副本数量）

--partitions 指定分区个数

--topic 指定所要创建主题的名称，比如test

replication-factor 表示该topic需要在不同的broker中保存几份，这里设置成1，表示在两个broker中保 存两份Partitions分区数。

1.3.2 查看topic

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

1.3.3 查看topic属性

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

1.3.4 消费消息

sh kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic test --from-beginning

1.3.5 发送消息

再开启一个终端窗口：

sh kafka-console-producer.sh --broker-list 127.0.0.1:9092 --topic test

输入：

消费端显示：

 1.4 kafka-topics.sh 使用方式

围绕创建、修改、删除以及查看等功能。

1.4.1 查看帮助--help

/bin目录下的每一个脚本工具，都有着众多的参数选项，不可能所有命令都记得住，这些脚本都可以使 用 --help 参数来打印列出其所需的参数信息。

sh kafka-topics.sh --help

Command must include exactly one action: --list, --describe, --create, --alter
or --delete
Option Description
------ -----------
--alter Alter the number of partitions,
replica assignment, and/or
configuration for the topic.
--config <String: name=value> A topic configuration override for the
topic being created or altered.The
following is a list of valid
configurations:
cleanup.policy
compression.type
delete.retention.ms
file.delete.delay.ms
flush.messages
flush.ms
follower.replication.throttled.
replicas
....省略

下面我们挑选其中使用最为频繁且重要的参数进行说明，以及其中一些坑进行标明。

1.4.2 副本数量不能大于broker的数量

kafka 创建主题的时候其副本数量不能大于broker的数量，否则创建主题 topic 失败.

sh kafka-topics.sh --create --zookeeper localhost:2181 -replication-factor 2 - -partitions 1 --topic test1

详细报错信息见下图

 注意：副本数量和分区数量的区别。

1.4.3 创建主题--create

创建主题时候，有3个参数是必填的，分别是 --partitions（分区数量）、 --topic（主题名） 、 -- replication-factor（复制系数）， 同时还需使用 --create 参数表明本次操作是想要创建一个主题操作。

sh kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1

--topic test1

返回： Created topic "test1".   此时主题 test1 就已经创建了。另外在创建主题的时候，还可以附加以下两个选项：–if-not-exists 和 -- if-exists . 第一个参数表明仅当该主题不存在时候，创建； 第二个参数表明当修改或删除这个主题时候， 仅在该主题存在的时候去执行操作。

1.4.4 查看broker上所有的主题 --list

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

1.4.5 查看指定主题 topic 的详细信息 --describe

该参数会将该主题的所有信息一一列出打印出来，比如分区数量、副本系数、领导者等待。

sh kafka-topics.sh --describe --zookeeper localhost:2181 --topic test1

 1.4.6 修改主题信息 --alter（增加主题分区数量）

 sh kafka-topics.sh --zookeeper localhost:2181 --topic test1 --alter --partitions 2

可以看到已经成功的将主题的分区数量从1修改为了2。如果去修改一个不存在的topic信息会怎么样？比如修改主题 test2，当前这主题是不存在的。

sh kafka-topics.sh --zookeeper localhost:2181 --topic test2 --alter --partitions 2

注意：不要使用 --alter 去尝试减少分区的数量，如果非要减少分区的数量，只能删除整个主题 topic， 然后重新创建。

1.4.7 删除主题 topic --delete

sh kafka-topics.sh --zookeeper localhost:2181 --delete --topic test1

日志信息提示，主题 test1已经被标记删除状态，但是若delete.topic.enable 没有设置为 true ， 则将不 会有任何作用。

启动生产者：sh kafka-console-producer.sh --broker-list 127.0.0.1:9092 --topic test1

启动消费者：sh kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic test1--from-beginning

发现此时还是可以发送消息和接收消息。

如果要支持能够删除主题的操作，则需要在 /bin 的同级目录 /config目录下的文件server.properties 中，修改配置delete.topic.enable=true（如果置为false，则kafka broker 是不允许删除主题的）。

 需要server.properties中设置delete.topic.enable=true否则只是标记删除或者直接重启。 重启kafka

停止：sh kafka-server-stop.sh -daemon ../config/server.properties

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

再次删除

sh kafka-topics.sh --zookeeper localhost:2181 --delete --topic test1