一、環境準備

jdk下載地址鏈接：jdk 1.8，提取碼: dv5h
zookeeper下載地址鏈接：zookeeper3.4.14 ，提取碼: 3dch
kafka下載地址鏈接：kafka2.12，提取碼: 61bc

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-amd64 export PATH=$PATH:$JAVA_HOME/bin

# 驗證 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.cfg cp zoo_sample.cfg zoo.cfg # 編輯zoo.cfg文件 vim zoo.cfg dataDir=/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.14 export PATH=$PATH:$ZOOKEEPER_PREFIX/bin export 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.2 export 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.sh Create, delete, describe, or change a topic. Option Description ------ ----------- --alter Alter the number of partitions, replica assignment, and/or configuration for the topic. --config <String: name=value> A topic configuration override for the topic being created or altered.The following is a list of valid configurations: cleanup.policy compression.type delete.retention.ms file.delete.delay.ms flush.messages flush.ms follower.replication.throttled. replicas index.interval.bytes leader.replication.throttled.replicas max.message.bytes message.format.version message.timestamp.difference.max.ms message.timestamp.type min.cleanable.dirty.ratio min.compaction.lag.ms min.insync.replicas preallocate retention.bytes retention.ms segment.bytes segment.index.bytes segment.jitter.ms segment.ms unclean.leader.election.enable See the Kafka documentation for full details on the topic configs. --create Create a new topic. --delete Delete a topic --delete-config <String: name> A topic configuration override to be removed for an existing topic (see the list of configurations under the --config option). --describe List details for the given topics. --disable-rack-aware Disable rack aware replica assignment --force Suppress console prompts --help Print usage information. --if-exists if set when altering or deleting topics, the action will only execute if the topic exists --if-not-exists if set when creating topics, the action will only execute if the topic does not already exist --list List all available topics. --partitions <Integer: # of partitions> The number of partitions for the topic being created or altered (WARNING: If partitions are increased for a topic that has a key, the partition logic or ordering of the messages will be affected --replica-assignment <String: A list of manual partition-to-broker broker_id_for_part1_replica1 : assignments for the topic being broker_id_for_part1_replica2 , created or altered. broker_id_for_part2_replica1 : broker_id_for_part2_replica2 , ...> --replication-factor <Integer: The replication factor for each replication factor> partition in the topic being created. --topic <String: topic> The topic to be create, alter or describe. Can also accept a regular expression except for --create option --topics-with-overrides if set when describing topics, only show topics that have overridden configs --unavailable-partitions if set when describing topics, only show partitions whose leader is not available --under-replicated-partitions if set when describing topics, only show under replicated partitions --zookeeper <String: urls> REQUIRED: The connection string for the zookeeper connection in the form host:port. Multiple URLS can be given 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用于生產消息

# 開啟生產者 [root@node1 ~]# kafka-console-producer.sh --topic topic_test --broker-list localhost:9020

2.4 具體操作

開啟消費者和生產者，生產并消費消息。

消費者，按照偏移量消費

消費者從頭消費，不按照偏移量消費

總結

以上是生活随笔為你收集整理的聊聊 Kafka： 在 Linux 环境上搭建 Kafka的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。