kafka是分布式消息队列，本文讲述其在centos（centos 7.5）下的安装。安装过程可以参考其官方文档https://kafka.apache.org/36/documentation.html

首先在官网 https://kafka.apache.org/downloads 下载Kafka二进制文件（官网的压缩包里面已经包含了Linux和windows的Kafka可执行文件了），选择目前比较新的3.6.1版本下载。

一、安装Java环境

 Kafka 是用 Java 开发的，因此需要安装java环境。从Kafka的官方文档 Apache Kafka 可以看到，Kafka3.6.1需要的java环境为Java8 或以上（不同版本的Kafka需要的Java环境版本也不一样）。

从Java官网Java Downloads | Oracle 下载 JDK（SUN公司提供的一套Java开发环境，简称JDK(JavaDevelopmentKit)，它是整个Java的核心，其中包括Java编译器、Java运行工具、Java文档生成工具、Java打包工具）

将java jdk拷贝到CentOS中，执行命令：

tar -zxvf jdk-21_linux-x64_bin.tar.gz
cd jdk-21.0.2
mkdir /usr/lib/jdk
cd ..
mv jdk-21.0.2/ /usr/lib/jdk

配置java环境变量，将环境变量配置在etc/profile或/etc/bashrc中

vi /etc/bashrc

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

#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

让环境变量生效

source /etc/bashrc

执行命令：

java -version

如果出现现版本号说明Java环境安装成功：

二、安装部署Kafka

然后我们来安装Kafka，将之前下载下来的Kafka二进制文件kafka_2.12-3.6.1.tgz拷贝到CentOS中，执行命令：

tar -zxvf kafka_2.12-3.6.1.tgz
mv kafka_2.12-3.6.1 kafka_3.6.1
cd kafka_3.6.1

可以看到目录结构如下：

.kafka3版本之后自带zookeeper。配置zookeeper的配置文件：

vi config/zookeeper.properties

设置快照存贮路径

配置Kafka的配置文件：

vi config/server.properties

设置日志文件的存贮路径：

启动zookeeper：

bin/zookeeper-server-start.sh -daemon config/zookeeper.properties

启动kafka:

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

通过netstat命令查看到9092端口被一个java进程占用，说明启动成功了

三、Kafka的基本操作

创建主题：

bin/kafka-topics.sh --create --topic quickstart-events --bootstrap-server localhost:9092

Kafka 中的消息以主题为单位进行归类，生产者负责将消息发送到特定的主题（发送到 Kafka 集群中的每一条消息都要指定一个主题），而消费者负责订阅主题并进行消费。

查看主题属性：

bin/kafka-topics.sh --describe --topic quickstart-events --bootstrap-server localhost:9092

新建两个终端，分别执行命令，通过生产者发送消息：

bin/kafka-console-producer.sh --topic quickstart-events --bootstrap-server localhost:9092

通过消费者消费消息：

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

然后就可以看到生产者和消费者进行通讯了：

四、遇到问题

Kafka生产者、消费者通讯时可能会报错：Error connecting to node xxxxx:9092，可以参考：《kafka发送数据报错： Error connecting to node xxxxx:9092 (id: 1 rack: null)java.net.UnknownHostExceptio》