Kafka在Mac下的安装与使用

安装kafka的原因

用户微服务登录后需要向广告微服务中发送用户登录的信息以获取用户画像（这个过程是异步的），故而在用户微服务中配置了kafka。配置的kafka的连接地址是测试环境的，为了避免给测试环境造成脏数据，因而我需要本地Mac安装kafka。

安装kafka

1. 我的网盘：链接: https://pan.baidu.com/s/1mS6MVBehe2Kko70Zdln-hA?pwd=62wr 提取码: 62wr

2. kafka的官网地址：https://www.apache.org/dyn/closer.cgi?path=/kafka/3.1.0/kafka_2.12-3.1.0.tgz
   

下载完成后，解压到当前目录中，或者移动到你想要的目录，比如

mv /Users/你的用户名/Downloads/kafka_2.12-3.1.0 /Users/你的用户名/kafka

进入kafka的bin目录，执行如下命令，查看bin目录下的文件，有下面要执行的文件。

cd ./kafka/bin
pwd 查看当前目录，确实为bin目录
ls  查看bin目录下的文件

你会看到zookeeper和kafka相关的文件，如下是具体操作zookeeper和kafka。

启动Zookeeper

打开新的terminal，进入Kafka的 bin目录，执行如下命令：

./zookeeper-server-start.sh ../config/zookeeper.properties &

zookeeper会打印一系列的日志，如下所示：

INFO Reading configuration from: ../config/zookeeper.properties (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2022-04-22 14:22:13,867] WARN ../config/zookeeper.properties is relative. Prepend ./ to indicate that you're sure! (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2022-04-22 14:22:13,879] INFO clientPortAddress is 0.0.0.0:2181 (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2022-04-22 14:22:13,879] INFO secureClientPort is not set (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2022-04-22 14:22:13,879] INFO observerMasterPort is not set (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2022-04-22 14:22:13,879] INFO metricsProvider.className is org.apache.zookeeper.metrics.impl.DefaultMetricsProvider (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2022-04-22 14:22:13,881] INFO autopurge.snapRetainCount set to 3 (org.apache.zookeeper.server.DatadirCleanupManager)
[2022-04-22 14:22:13,881] INFO autopurge.purgeInterval set to 0 (org.apache.zookeeper.server.DatadirCleanupManager)
[2022-04-22 14:22:13,881] INFO Purge task is not scheduled. (org.apache.zookeeper.server.DatadirCleanupManager)
[2022-04-22 14:22:13,881] WARN Either no config or no quorum defined in config, running in standalone mode (org.apache.zookeeper.server.quorum.QuorumPeerMain)
[2022-04-22 14:22:13,884] INFO Log4j 1.2 jmx support found and enabled. (org.apache.zookeeper.jmx.ManagedUtil)
[2022-04-22 14:22:13,896] INFO Reading configuration from: ../config/zookeeper.properties (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2022-04-22 14:22:13,896] WARN ../config/zookeeper.properties is relative. Prepend ./ to indicate that you're sure! (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2022-04-22 14:22:13,896] INFO clientPortAddress is 0.0.0.0:2181 (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2022-04-22 14:22:13,897] INFO secureClientPort is not set (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2022-04-22 14:22:13,897] INFO observerMasterPort is not set (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2022-04-22 14:22:13,897] INFO metricsProvider.className is org.apache.zookeeper.metrics.impl.DefaultMetricsProvider (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2022-04-22 14:22:13,897] INFO Starting server (org.apache.zookeeper.server.ZooKeeperServerMain)
[2022-04-22 14:22:13,912] INFO ServerMetrics initialized with provider org.apache.zookeeper.metrics.impl.DefaultMetricsProvider@60addb54 (org.apache.zookeeper.server.ServerMetrics)
[2022-04-22 14:22:13,916] INFO zookeeper.snapshot.trust.empty : false (org.apache.zookeeper.server.persistence.FileTxnSnapLog)
[2022-04-22 14:22:13,930] INFO  (org.apache.zookeeper.server.ZooKeeperServer)
[2022-04-22 14:22:13,930] INFO   ______                  _                                           (org.apache.zookeeper.server.ZooKeeperServer)
[2022-04-22 14:22:13,930] INFO  |___  /                 | |                                          (org.apache.zookeeper.server.ZooKeeperServer)
[2022-04-22 14:22:13,930] INFO     / /    ___     ___   | | __   ___    ___   _ __     ___   _ __    (org.apache.zookeeper.server.ZooKeeperServer)
[2022-04-22 14:22:13,930] INFO    / /    / _ \   / _ \  | |/ /  / _ \  / _ \ | '_ \   / _ \ | '__| (org.apache.zookeeper.server.ZooKeeperServer)
[2022-04-22 14:22:13,931] INFO   / /__  | (_) | | (_) | |   <  |  __/ |  __/ | |_) | |  __/ | |     (org.apache.zookeeper.server.ZooKeeperServer)
[2022-04-22 14:22:13,931] INFO  /_____|  \___/   \___/  |_|\_\  \___|  \___| | .__/   \___| |_| (org.apache.zookeeper.server.ZooKeeperServer)
[2022-04-22 14:22:13,931] INFO                                               | |                      (org.apache.zookeeper.server.ZooKeeperServer)
[2022-04-22 14:22:13,931] INFO                                               |_|                      (org.apache.zookeeper.server.ZooKeeperServer)
[2022-04-22 14:22:13,931] INFO  (org.apache.zookeeper.server.ZooKeeperServer)
[2022-04-22 14:22:13,933] INFO Server environment:zookeeper.version=3.6.3--6401e4ad2087061bc6b9f80dec2d69f2e3c8660a, built on 04/08/2021 16:35 GMT (org.apache.zookeeper.server.ZooKeeperServer)
......

启动Kafka

打开新的terminal，进入Kafka的 bin目录，执行如下命令：

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

你会看到如下一系列日志，直到看到这句话INFO [KafkaServer id=0] started (kafka.server.KafkaServer)就说明启动成功了

INFO Registered kafka:type=kafka.Log4jController MBean (kafka.utils.Log4jControllerRegistration$)
[2022-04-22 14:23:55,309] INFO Setting -D jdk.tls.rejectClientInitiatedRenegotiation=true to disable client-initiated TLS renegotiation (org.apache.zookeeper.common.X509Util)
[2022-04-22 14:23:55,384] INFO Registered signal handlers for TERM, INT, HUP (org.apache.kafka.common.utils.LoggingSignalHandler)
[2022-04-22 14:23:55,388] INFO starting (kafka.server.KafkaServer)
[2022-04-22 14:23:55,388] INFO Connecting to zookeeper on localhost:2181 (kafka.server.KafkaServer)
[2022-04-22 14:23:55,411] INFO [ZooKeeperClient Kafka server] Initializing a new session to localhost:2181. (kafka.zookeeper.ZooKeeperClient)
[2022-04-22 14:23:55,417] INFO Client environment:zookeeper.version=3.6.3--6401e4ad2087061bc6b9f80dec2d69f2e3c8660a, built on 04/08/2021 16:35 GMT (org.apache.zookeeper.ZooKeeper)
[2022-04-22 14:23:55,417] INFO Client environment:host.name=localhost (org.apache.zookeeper.ZooKeeper)
[2022-04-22 14:23:55,417] INFO Client environment:java.version=1.8.0_211 (org.apache.zookeeper.ZooKeeper)
[2022-04-22 14:23:55,417] INFO Client environment:java.vendor=Oracle Corporation (org.apache.zookeeper.ZooKeeper)
[2022-04-22 14:23:55,417] INFO Client environment:java.home=/Library/Internet Plug-Ins/JavaAppletPlugin.plugin/Contents/Home (org.apache.zookeeper.ZooKeeper)
。。。

创建topic

开启新的terminal，进入kafka的bin目录，执行如下命令

./kafka-topics.sh --bootstrap-server localhost:9092 --create --topic testInfoTopic --partitions 2 --replication-factor 1

你会看到如上图所示的信息，就说明创建topic成功。

查看topic

基于上一个terminal的kafka的bin目录，执行如下命令

./kafka-topics.sh --list --bootstrap-server localhost:9092

你会看到所创建的topic：testInfoTopic

生产数据

基于上一个terminal的kafka的bin目录，执行如下命令：

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

先不输入数据。

消费数据

开启新的terminal，进入kafka的bin目录中，执行如下命令

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic testInfoTopic

你会看到，此时没有数据。

在生产者的terminal中发送消息:

在消费者的terminal中消费消息：

综上所看，消息成功发送，也被成功消费。

关闭zookeeper

开启新的terminal，进入kafka的bin目录，执行如下命令

sh zookeeper-server-stop.sh

再次执行这个命令，你会看到No zookeeper server to stop，说明zookeeper已关闭。

但是执行命令ps -eaf | grep zookeeper，仍看到zookeeper的进程，说明还有其子节点没有关闭。此为kafka没有关闭，即70267，于是去关闭kafka.

关闭kafka

在上面的terminal中，执行如下命令关闭kafka

sh kafka-server-stop.sh

再次执行该命令，你会看到No kafka server to stop。
执行命令ps -eaf | grep zookeeper，没看到70267的kafka节点进程，说明kafka已关闭。
进而执行命令ps -eaf | grep kafka ，没有kafka的进程，再次说明kafka已关闭。

因而，即便zookeeper关闭了，如果其下节点没有关闭，节点进程仍旧存在。因而，zookeeper和kafka都要关闭。

测试

启动用户微服务，其kafka的topic是UserLoginInfo。

使用postman调用登录接口，同时，在terminal启动kafka和zookeeper服务，再开启一新terminal，并执行命令：./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic UserLoginInfo ，得到如下消息：

说明kafka和zookeeper是生效的。

注意事项：./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic UserLoginInfo 在接口请求之前要执行。