1、安装zookeeper

因为zookeeper 与kafka 存在对应的版本，选择不当，将无法使用，所以两者都使用最新版本

下载地址：https://zookeeper.apache.org/releases.html 

目录

下载并解压ZooKeeper软件压缩包后，可以看到zk包含以下的文件和目录：

图1：ZooKeeper软件的文件和目录

• • bin目录
    zk的可执行脚本目录，包括zk服务进程，zk客户端，等脚本。其中，.sh是Linux环境下的脚本，.cmd是Windows环境下的脚本。
  • conf目录
    配置文件目录。zoo_sample.cfg为样例配置文件，需要修改为自己的名称，一般为zoo.cfg。log4j.properties为日志配置文件。
  • lib
    zk依赖的包。
  • contrib目录
    一些用于操作zk的工具包。
  • recipes目录
    zk某些用法的代码示例

运行配置

上面提到，conf目录下提供了配置的样例zoo_sample.cfg，要将zk运行起来，需要将其名称修改为zoo.cfg。
打开zoo.cfg，可以看到默认的一些配置。

• • tickTime
    时长单位为毫秒，为zk使用的基本时间度量单位。例如，1 * tickTime是客户端与zk服务端的心跳时间，2 * tickTime是客户端会话的超时时间。
    tickTime的默认值为2000毫秒，更低的tickTime值可以更快地发现超时问题，但也会导致更高的网络流量（心跳消息）和更高的CPU使用率（会话的跟踪处理）。
  • clientPort
    zk服务进程监听的TCP端口，默认情况下，服务端会监听2181端口。
  • dataDir
    无默认配置，必须配置，用于配置存储快照文件的目录。如果没有配置dataLogDir，那么事务日志也会存储在此目录。

配置环境变量 vim ~/.bash_profile 

export ZK_HOME=/Users/yyj/big_data/zookeeper
export KAFKA_HOME=/Users/yyj/big_data/kafka_2.12-2.5.0
export PATH=$PATH:$ANDROID_HOME/platform-tools:${KAFKA_HOME}/bin:${ZK_HOME}/bin

启动 进入bin目录，执行命令

zkServer.sh start

安装Kafka

下载 http://kafka.apache.org/downloads

解压

tar -xzf kafka_2.12-2.5.0.tgz

注意，kafka_2.12-2.5.0.tgz版本是已经编译好的版本，解压就能使用。

 配置server.properties

默认配置 advertised.listeners=PLAINTEXT://:your.host.name:9092 修改为 advertised.listeners=PLAINTEXT://:ip:9092

ip为服务器ip。

hostname和端口是用来建议给生产者和消费者使用的，如果没有设置，将会使用listeners的配置，如果listeners也没有配置，将使用java.net.InetAddress.getCanonicalHostName()来获取这个hostname和port，对于ipv4，基本就是localhost了。

"PLAINTEXT"表示协议，可选的值有PLAINTEXT和SSL，hostname可以指定IP地址，也可以用"0.0.0.0"表示对所有的网络接口有效，如果hostname为空表示只对默认的网络接口有效。也就是说如果你没有配置advertised.listeners，就使用listeners的配置通告给消息的生产者和消费者，这个过程是在生产者和消费者获取源数据(metadata)。

配置环境变量

export ZK_HOME=/Users/yyj/big_data/zookeeper
export KAFKA_HOME=/Users/yyj/big_data/kafka_2.12-2.5.0
export PATH=$PATH:$ANDROID_HOME/platform-tools:${KAFKA_HOME}/bin:${ZK_HOME}/bin

启动Kafka

 启动ZooKeeper

zkServer.sh start

注意，需要先启动ZooKeeper再启动kafka，不然会报错。如下图：

启动kafka

kafka-server-start.sh /Users/yyj/big_data/kafka_2.12-2.5.0/config/server.properties

启动Kafka Broker后，在ZooKeeper终端上键入命令 jps,效果如下：

创建topic

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

其中demo为创建的topic名称。

如上图，创建了一个名为 demo 的主题，其中包含一个分区和一个副本因子。 创建成功之后会输出： Created topic "demo".

如上图，创建主题后，系统会在config / server.properties文件中的"/ tmp / kafka-logs /"中指定的创建主题的日志。

 查询topic列表

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

 查看topic信息

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

 删除topic

bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic demo

启动生产者

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

从上面的语法，生产者命令行客户端需要两个主要参数 -

代理列表 - 我们要发送邮件的代理列表。 在这种情况下，我们只有一个代理。 Config / server.properties文件包含代理端口ID，因为我们知道我们的代理正在侦听端口9092，因此您可以直接指定它。主题名称:demo。

 启动消费者

为了方便测试，另启一个sheel窗口 这样效果更明显。需要注意的是旧版本和新版本的命令是不一样的

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

报错提示： zookeeper is not a recognized option

发现在启动的时候说使用 --zookeeper是一个过时的方法，最新的版本中命令如下：

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic demo --from-beginning

可以开启两个终端，一个发送消息，一个接受消息。效果如下：

配置启动 关闭shell 脚本

start_all.sh 

#!/bin/bash

# 启动zk
zkServer.sh start

# 启动kafka
nohup kafka-server-start.sh /Users/yyj/big_data/kafka_2.12-2.5.0/config/server.properties >> /Users/yyj/big_data/kafka.log &

stop_all.sh

#!/bin/bash

# 关闭kafka
kafka-server-stop.sh /Users/yyj/big_data/kafka_2.12-2.5.0/config/server.properties

# 关闭zk
zkServer.sh stop