一.下载安装包官方下载地址Downloads | Apache Kafka之后通过winscp将文件放到指定目录下我放到/opt下二.解压cd /opttar –zxvf kafka_2.13-4.3.0.tgz此时可以看到kafka目录下文件列表如下三.升级jdk版本#查看java版本如果显示 1.8.x 或 java version 1.8.0_xxx说明是 JDK 8java –version注如果报错则说明配置的环境变量没有生效则每次登录shell都需输入source /etc/profile也可一劳永逸解决方式写入到环境变量同时可解决退出会话窗口kafka服务就停止以及无法使用systemctl status kafka的问题实现方式如下vim /root/.bashrc添加如下内容[Unit] DescriptionApache Kafka Server Afternetwork.target [Service] Typesimple Userroot # 关键显式指定 JAVA_HOME EnvironmentJAVA_HOME/opt/jdk17 WorkingDirectory/opt/kafka_2.13-4.3.0 ExecStart/opt/kafka_2.13-4.3.0/bin/kafka-server-start.sh /opt/kafka_2.13-4.3.0/config/server.properties ExecStop/opt/kafka_2.13-4.3.0/bin/kafka-server-stop.sh Restarton-failure [Install] WantedBymulti-user.target升级方式1sudo yum install java-17-openjdk -ysudo yum install java-17-openjdk-devel -y # 开发工具包可选升级方式2如果方式1失败可用方式21. 进入系统临时文件夹cd /tmp2. 下载亚马逊提供的 JDK 17 安装包这个过程很快sudo wget https://corretto.aws/downloads/latest/amazon-corretto-17-x64-linux-jdk.rpm3. 使用 yum 本地安装这个包sudo yum localinstall -y amazon-corretto-17-x64-linux-jdk.rpm4.再次检查jdk版本这时正常四.修改配置首先在/opt/kafka_2.13-4.3.0目录下创建data目录备用mkdir data因为不使用zookeeper所以来到kafka装根目录的下的configcd /opt/kafka_2.13-4.3.0/configvi server.properties我这里只修改了日志路径log.dirs/opt/kafka_2.13-4.3.0/data五.生成集群进入bin目录下输入命令生成集群id./kafka-storage.sh random-uuid得到uuidLC4o1MlETqC8q0vltsaywQ注如果jdk版本过低会报错需要升级到17及以上版本具体见第三节六.格式化目录发现有如下报错可以在最后加上 --standalone这样Kafka 就知道你是在单机模式下运行不会再问关于集群控制器配置的问题。此时格式化成功格式化操作完成之后会发现在我们定义的log.dirs目录下多出一个meta.properties文件。meta.properties文件中存储了当前的kafka节点的id(node.id)当前节点属于哪个集群cluster.id七.启动 Kafka1.格式化成功后启动 Kafka./kafka-server-start.sh -daemon ../config/server.properties //运行在后台./kafka-server-start.sh ../config/server.properties //运行在前台此时会生成大量的日志为正常现象2.验证 Kafka 是否启动成功#查看运行状态systemctl status kafka也可以用以下命令快速确认Kafka运行是否正常# 检查进程是否有kafkajps | grep Kafka# 检查端口默认 9092netstat -tlnp | grep 9092八.kafka使用Kafka 的核心概念其实很好理解你可以把它想象成一个邮政系统①Producer (生产者)发信人②Topic (主题)收件地址③Broker (服务器)邮局④Consumer (消费者)收信人打开两个新的终端窗口开始第一步创建主题在新窗口1执行先创建一个名为test的主题。/opt/kafka_2.13-4.3.0/bin/kafka-topics.sh --create --topic test --bootstrap-server localhost:9092第二步扮演“生产者”发消息在窗口1运行生产脚本。输入任意文字按回车发送/opt/kafka_2.13-4.3.0/bin/kafka-console-producer.sh --topic test --bootstrap-server localhost:9092屏幕出现提示符后输入hello kafka或任何内容试试。第三步扮演“消费者”收消息在窗口2新窗口运行消费者脚本/opt/kafka_2.13-4.3.0/bin/kafka-console-consumer.sh --topic test --from-beginning --bootstrap-server localhost:9092此时你应该能立刻看到刚才在窗口1输入的消息。试试在窗口1继续打字窗口2会实时显示。注--from-beginning 这个参数的意思是从主题Topic里最早的消息开始读取而不仅仅是新产生的消息。如果不加这个参数消费者默认只会读取启动之后才产生的新消息之前的历史消息会被忽略。九.常用命令#设置开机自启如果希望 Kafka 随系统启动systemctl enable kafka#查看完整日志如果发现任何可疑信息journalctl -u kafka -n 50 --no-pager#常用管理命令systemctl start kafka # 启动systemctl stop kafka # 停止systemctl restart kafka # 重启systemctl status kafka # 查看状态或bin/kafka-server-stop.shbin/kafka-server-start.sh config/server.properties //加-daemon则放在后台进行#创建一个测试 topic./kafka-topics.sh --create --topic test --bootstrap-server localhost:9092#查看所有主题bin/kafka-topics.sh --list --bootstrap-server localhost:9092#查看主题详情分区、副本状态bin/kafka-topics.sh --describe --topic test --bootstrap-server localhost:9092#删除主题清理测试数据bin/kafka-topics.sh --delete --topic test --bootstrap-server localhost:9092#查看topic有没有消息bin/kafka-run-class.sh org.apache.kafka.tools.GetOffsetShell --bootstrap-server localhost:9092 --topic CFW_ops --time -1#查看消费进度检查有没有消息积压bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --all-groups --list