Docker安装启动Kafka

1、下载Kafka和Zookeeper镜像文件

docker pull wurstmeister/kafka
docker pull wurstmeister/zookeeper

2、先运行zk

docker run -d --name zookeeper -p 2181:2181 -t wurstmeister/zookeeper

3、再运行kafka

docker run --name kafka01 \
-p 9092:9092 \
-e KAFKA_BROKER_ID=0 \
-e KAFKA_ZOOKEEPER_CONNECT=172.19.0.241:2181 \
-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://172.19.0.241:9092 \
-e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 \
-d  wurstmeister/kafka

4、查看容器是否正常运行

docker ps

5、进行生产消费测试

docker exec -it kafka01 bash

cd /opt/kafka_2.13-2.6.0/bin/

6、运行kafka生产者发送消息

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

7、运行kafka消费者接收消息

/opt/kafka_2.13-2.6.0/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

在docker上启动kafka,是因为最近我们开发的程序需要部署到k8s，在k8s上连接kafka，连接失败，因此我先搭建个环境，以供后续还原问题，解决问题。

一、docker上启动kafka

1.下载zookeeper镜像

docker pull wurstmeister/zookeeper

2.下载kafka镜像

docker pull wurstmeister/kafka

3.启动zookeeper

docker run -d --name zookeeper -p 2181:2181  -t wurstmeister/zookeeper

4.启动kafka

docker run -d --name kafka --publish 9092:9092 --link zookeeper:zookeeper -e KAFKA_BROKER_ID=1  -e HOST_IP=192.168.65.7  -e KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.65.7:9092  -e KAFKA_ADVERTISED_HOST_NAME=192.168.65.7 -e KAFKA_ADVERTISED_PORT=9082  --restart=always  -t  wurstmeister/kafka

部分参数含义：

KAFKA_BROKER_ID=1 kafka在集群中的唯一标识
KAFKA_ZOOKEEPER_CONNECT=zookeeper:12181 监听zookeeper的地址 
KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.65.7:19092 kafka服务器的地址

5.进入kafka容器

1.先查看kafka镜像ID:
docker ps -a

2.进入容器
docker exec -it ${CONTAINER ID} /bin/bash
3.进入kafka的bin目录下
cd /opt/kafka_2.18/bin
4.创建topic
./kafka-topics.sh --create --zookeeper zookeeper:2181 --replication-factor 1 --partitions 1 --topic mykafka
5.运行生产者
./kafka-console-producer.sh --broker-list localhost:9092 --topic mykafka

6.消费topic
./kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --from-beginning --topic mykafka

6.问题记录

1.当第一次启动kafka时报以下错误，连接超时的错，我报这错是因为我启动命令错了：
 ERROR Fatal error during KafkaServer startup. Prepare to shutdown (kafka.server.KafkaServer)
kafka.zookeeper.ZooKeeperClientTimeoutException: Timed out waiting for connection while in state: CONNECTING At kafka.zookeeper.ZooKeeperClient.waitUntilConnected(ZooKeeperClient.scala:271)
        at kafka.zookeeper.ZooKeeperClient.<init>(ZooKeeperClient.scala:125)
        at kafka.zk.KafkaZkClient$.apply(KafkaZkClient.scala:1948)
        at kafka.server.KafkaServer.createZkClient$1(KafkaServer.scala:431)
        at kafka.server.KafkaServer.initZkClient(KafkaServer.scala:456)
        at kafka.server.KafkaServer.startup(KafkaServer.scala:191)
        at kafka.Kafka$.main(Kafka.scala:109)
        at kafka.Kafka.main(Kafka.scala)
[2022-01-20 07:33:19,615] INFO shutting down (kafka.server.KafkaServer)
[2022-01-20 07:33:19,645] INFO App info kafka.server for 0 unregistered (org.apache.kafka.common.utils.AppInfoParser)
[2022-01-20 07:33:19,646] INFO shut down completed (kafka.server.KafkaServer)
[2022-01-20 07:33:19,647] ERROR Exiting Kafka. (kafka.Kafka$)
[2022-01-20 07:33:19,657] INFO shutting down (kafka.server.KafkaServer)
INFO Opening socket connection to server 192.168.65.7/192.168.65.7:2181. Will not attempt to authenticate using SASL (unknown error) (org.apache.zookeeper.ClientCnxn)
报错的启动命令：
docker run -d --name kafka --publish 9092:9092 --link zookeeper --env KAFKA_ZOOKEEPER_CONNECT=192.168.65.7:2181 --env KAFKA_ADVERTISED_HOST_NAME=192.168.65.7 --env KAFKA_ADVERTISED_PORT=9092 wurstmeister/kafka

总结

启动kafka遇到了很多问题，一直以为自己是网络不通，防火墙没开之类的，最后发现一直是自己的运行命令不正确，菜鸟的我，还需继续努力呀。

安装kafka

brew install kafka

通过brew安装，会默认直接安装最新的版本。

因为我的电脑系统版本比较低，所以brew不支持最新的kafka版本。本文介绍的是通过官网下载安装包的方式。

下载kafka

下载地址：https://kafka.apache.org/downloads.html

注意：一定不要下成源文件了，是不带有-src 的文件

解压(以下是本机的对应目录)

同brew安装的kafka包和配置文件的路径分别为

启动kafka

首先需要启动zookeeper

##进入到自己相对应的目录
cd Downloads/zookeeper-3.4.6
bin/zkServer.sh start

然后在启动kafka

bin/kafka-server-start /usr/local/etc/kafka/server.properties

在启动kakfa的时候会出现下列错误

ERROR Fatal error during KafkaServer startup. Prepare to shutdown (kafka.server.KafkaServer)
kafka.common.InconsistentClusterIdException: The Cluster ID EcpP-vcMS9-LwdxE8WrF8A doesn't match stored clusterId Some(IdtIKWu6SzezUGXvPRChtg) in meta.properties. The broker is trying to join the wrong cluster. Configured zookeeper.connect may be wrong.
	at kafka.server.KafkaServer.startup(KafkaServer.scala:223)
	at kafka.Kafka$.main(Kafka.scala:109)
	at kafka.Kafka.main(Kafka.scala)

解决方式：
当出现这个错误时，从以下两个角度排查：
（1）zk集群是否运行正常；
（2）kafka的logs.dir数据目录是否为空（在初次安装时可直接删除掉该目录，但已有生产数据不要这么操作）

创建topic

kafka-topics --create --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1 --topic test

显示topic

kafka-topics --list

创建生产者

新建一个窗口

 kafka-console-producer --bootstrap-server localhost:9092 --topic test

创建消费者

kafka-console-consumer --bootstrap-server localhost:9092  --topic test