在Linux环境下安装Kafka 在Linux环境下安装Kafka

二、生产与消费

2.1 kafka-topics.sh 用于管理主题
2.2 kafka-console-consumer.sh用于消费消息
2.3 kafka-console-producer.sh用于生产消息
2.4 具体操作

一、环境准备 jdk下载地址链接：下载地址
zookeeper下载地址链接：下载地址

kafka下载地址链接：下载地址
1.1 Java环境为前提
1.1.1 上传jdk-8u261-linux-x64.rpm到服务器并安装

# 安装命令rpm -ivh jdk-8u261-linux-x64.rpm

文章图片

1.1.2 配置环境变量

# 编辑配置文件,jdk的bin目录到/etc/profile文件，对所有用户的shell有效vim /etc/profile# 生效source /etc/profile

文章图片

export JAVA_HOME=/usr/java/jdk1.8.0_261-amd64export PATH=$PATH:$JAVA_HOME/bin

【在Linux环境下安装Kafka】

文章图片

# 验证java -version

文章图片

至此，jdk安装成功。
1.2 Zookeeper的安装配置
1.2.1 上传zookeeper-3.4.14.tar.gz到服务器，解压到/opt

# 解压zk到指定目录tar -zxf zookeeper-3.4.14.tar.gz -C /opt

文章图片

1.2.2 修改Zookeeper保存数据的目录，dataDir

# 进入conf配置目录cd /opt/zookeeper-3.4.14/conf# 复制zoo_sample.cfg命名为zoo.cfgcp zoo_sample.cfg zoo.cfg# 编辑zoo.cfg文件vim zoo.cfgdataDir=/var/riemann/zookeeper/data

文章图片

文章图片

1.2.3 编辑/etc/profile，使配置生效

设置环境变量ZOO_LOG_DIR，指定Zookeeper保存日志的位置；
ZOOKEEPER_PREFIX指向Zookeeper的解压目录；
将Zookeeper的bin目录添加到PATH中：

文章图片

export ZOOKEEPER_PREFIX=/opt/zookeeper-3.4.14export PATH=$PATH:$ZOOKEEPER_PREFIX/binexport ZOO_LOG_DIR=/var/riemann/zookeeper/log

配置完以后再生效配置：

source /etc/profile

1.2.4 启动Zookeeper，确认Zookeeper的状态

zkServer.sh start

文章图片

至此，zookeeper安装成功。
1.3 Kafka 的安装与配置
1.3.1 上传kafka_2.12-1.0.2.tgz到服务器并解压

tar -zxf kafka_2.12-1.0.2.tgz -C /opt

文章图片

1.3.2 配置环境变量并生效

vim /etc/profile

export KAFKA=/opt/kafka_2.12-1.0.2export PATH=$PATH:$KAFKA/bin

source /etc/profile

1.3.3 配置/opt/kafka_2.12-1.0.2/config中的server.properties文件

vi/opt/kafka_2.12-1.0.2/config/server.properties

文章图片

Kafka连接Zookeeper的地址，此处使用本地启动的Zookeeper实例
连接地址是localhost:2181
后面的 myKafka 是Kafka在Zookeeper中的根节点路径

文章图片

配置kafka存储持久化数据目录

文章图片

log.dirs=/var/riemann/kafka/kafka-logs

创建上述持久化数据目录

mkdir -p /var/riemann/kafka/kafka-logs

1.4 启动Kafka
进入Kafka安装的根目录，执行如下命令：

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

文章图片

启动成功，可以看到控制台输出的最后一行的started状态：此时kafka安装成功。

文章图片

1.5 重新开一个窗口，查看Zookeeper的节点

1.6 此时Kafka是前台模式启动，要停止，使用Ctrl+C
如果要后台启动，使用命令：

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

查看Kafka的后台进程：

ps -ef | grep kafka

文章图片

停止后台运行的Kafka：

kafka-server-stop.sh

二、生产与消费查看zookeeper状态，zookeeper启动成功，再启动kafka。

文章图片

2.1 kafka-topics.sh 用于管理主题
查看命令的帮助信息

[root@master1 bin]# kafka-topics.shCreate, delete, describe, or change a topic.OptionDescription-------------------alterAlter the number of partitions,replica assignment, and/orconfiguration for the topic.--configA topic configuration override for the topic being created or altered.Thefollowing is a list of validconfigurations:cleanup.policycompression.typedelete.retention.msfile.delete.delay.msflush.messagesflush.msfollower.replication.throttled.replicasindex.interval.bytesleader.replication.throttled.replicas max.message.bytesmessage.format.versionmessage.timestamp.difference.max.msmessage.timestamp.typemin.cleanable.dirty.ratiomin.compaction.lag.msmin.insync.replicaspreallocateretention.bytesretention.mssegment.bytessegment.index.bytessegment.jitter.mssegment.msunclean.leader.election.enableSee the Kafka documentation for fulldetails on the topic configs.--createCreate a new topic.--deleteDelete a topic--delete-configA topic configuration override to beremoved for an existing topic (seethe list of configurations under the --config option).--describeList details for the given topics.--disable-rack-awareDisable rack aware replica assignment--forceSuppress console prompts--helpPrint usage information.--if-existsif set when altering or deletingtopics, the action will only execute if the topic exists--if-not-existsif set when creating topics, theaction will only execute if thetopic does not already exist--listList all available topics.--partitions The number of partitions for the topic being created or altered (WARNING:If partitions are increased for atopic that has a key, the partitionlogic or ordering of the messageswill be affected--replica-assignment--replication-factor partition in the topic being created.--topicThe topic to be create, alter ordescribe. Can also accept a regularexpression except for --create option--topics-with-overridesif set when describing topics, onlyshow topics that have overriddenconfigs--unavailable-partitionsif set when describing topics, onlyshow partitions whose leader is notavailable--under-replicated-partitionsif set when describing topics, onlyshow under replicated partitions--zookeeperREQUIRED: The connection string forthe zookeeper connection in the form host:port. Multiple URLS can begiven to allow fail-over.[root@master1 bin]#

# 列出现有的主题[root@master1 ~]# kafka-topics.sh --list --zookeeper localhost:2181/myKafka# 创建主题，该主题包含一个分区，该分区为Leader分区，它没有Follower分区副本。[root@master1 ~]# kafka-topics.sh --zookeeper localhost:2181/myKafka --create --topic topic_test --partitions 1 --replication-factor 1# 查看分区信息[root@master1 ~]# kafka-topics.sh --zookeeper localhost:2181/myKafka --list# 查看指定主题的详细信息[root@master1 ~]# kafka-topics.sh --zookeeper localhost:2181/myKafka --describe --topic topic_test # 删除指定主题[root@master1 ~]# kafka-topics.sh --zookeeper localhost:2181/myKafka --delete --topic topic_test

列出现有主题，创建主题，该主题包含一个分区，该分区为Leader分区，它没有Follower分区副本。

文章图片

查看指定主题的详细信息

文章图片

创建主题，该主题包含多个分区

多个分区：横向扩展
多个副本：高可用

2.2 kafka-console-consumer.sh用于消费消息

# 开启消费者[root@node1 ~]# kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topic_test # 开启消费者方式二，从头消费，不按照偏移量消费[root@node1 ~]# kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topic_test --from-beginning

2.3 kafka-console-producer.sh用于生产消息