溫馨提示×

您好,登錄后才能下訂單哦!

密碼登錄×
登錄注冊(cè)×
其他方式登錄
點(diǎn)擊 登錄注冊(cè) 即表示同意《億速云用戶服務(wù)條款》

ubuntu16.04系統(tǒng)中怎么安裝kafka

發(fā)布時(shí)間:2021-06-15 13:53:25 來(lái)源:億速云 閱讀:235 作者:Leah 欄目:大數(shù)據(jù)

ubuntu16.04系統(tǒng)中怎么安裝kafka,很多新手對(duì)此不是很清楚,為了幫助大家解決這個(gè)難題,下面小編將為大家詳細(xì)講解,有這方面需求的人可以來(lái)學(xué)習(xí)下,希望你能有所收獲。

下載

wget http://mirror-hk.koddos.net/apache/kafka/2.3.0/kafka_2.12-2.3.0.tgz

安裝

tar zxvf kafka_2.12-2.3.0.tgz
cd kafka_2.12-2.3.0/
vim config/server.properties

配置

# 通用配置
# kafka數(shù)據(jù)目錄
log.dirs=/data/kafka

# zookeeeper
zookeeper.connect=kafka-node1:2181,kafka-node2:2181,kafka-node3:2181

# 節(jié)點(diǎn)配置
# 節(jié)點(diǎn)1
broker.id=0
#listeners=PLAINTEXT://:9092
listeners=PLAINTEXT://10.13.6.1:9092

# 節(jié)點(diǎn)2
broker.id=1
#listeners=PLAINTEXT://:9092
listeners=PLAINTEXT://10.13.6.2:9092

# 節(jié)點(diǎn)3
broker.id=2
#listeners=PLAINTEXT://:9092
listeners=PLAINTEXT://10.13.6.3:9092

啟動(dòng)

#進(jìn)入kafka根目錄
cd /app/kafka_2.12-2.3.0/
#啟動(dòng)
bin/kafka-server-start.sh -daemon config/server.properties

#啟動(dòng)成功輸出示例(最后幾行)
[2019-09-11 11:14:13,403] INFO [ProducerId Manager 0]: Acquired new producerId block (brokerId:0,blockStartProducerId:0,blockEndProducerId:999) by writing to Zk with path version 1 (kafka.coordinator.transaction.ProducerIdManager)
[2019-09-11 11:14:13,423] INFO [TransactionCoordinator id=0] Starting up. (kafka.coordinator.transaction.TransactionCoordinator)
[2019-09-11 11:14:13,424] INFO [Transaction Marker Channel Manager 0]: Starting (kafka.coordinator.transaction.TransactionMarkerChannelManager)
[2019-09-11 11:14:13,424] INFO [TransactionCoordinator id=0] Startup complete. (kafka.coordinator.transaction.TransactionCoordinator)
[2019-09-11 11:14:13,459] INFO [/config/changes-event-process-thread]: Starting (kafka.common.ZkNodeChangeNotificationListener$ChangeEventProcessThread)
[2019-09-11 11:14:13,479] INFO [SocketServer brokerId=0] Started data-plane processors for 1 acceptors (kafka.network.SocketServer)
[2019-09-11 11:14:13,485] INFO Kafka version: 2.3.0 (org.apache.kafka.common.utils.AppInfoParser)
[2019-09-11 11:14:13,485] INFO Kafka commitId: fc1aaa116b661c8a (org.apache.kafka.common.utils.AppInfoParser)
[2019-09-11 11:14:13,485] INFO Kafka startTimeMs: 1568171653480 (org.apache.kafka.common.utils.AppInfoParser)
[2019-09-11 11:14:13,487] INFO [KafkaServer id=0] started (kafka.server.KafkaServer)

使用

1、創(chuàng)建Topic
在kafka-node1(Broker)上創(chuàng)建測(cè)試Tpoic:test-ken-io,這里我們指定了3個(gè)副本、1個(gè)分區(qū)
bin/kafka-topics.sh --create --bootstrap-server kafka-node1:9092 --replication-factor 3 --partitions 1 --topic test-ken-io
Topic在kafka-node1上創(chuàng)建后也會(huì)同步到集群中另外兩個(gè)Broker:kafka-node2、kafka-node3

2、查看Topic
我們可以通過(guò)命令列出指定Broker的
bin/kafka-topics.sh --list --bootstrap-server kafka-node1:9092

3、發(fā)送消息
這里我們向Broker(id=0)的Topic=test-ken-io發(fā)送消息
bin/kafka-console-producer.sh --broker-list  kafka-node1:9092  --topic test-ken-io

#消息內(nèi)容
> test by ken.io

4、消費(fèi)消息
在kafka-node2上消費(fèi)Broker03的消息
bin/kafka-console-consumer.sh --bootstrap-server kafka-node3:9092 --topic test-ken-io --from-beginning

在Kafka03上消費(fèi)Broker02的消息
bin/kafka-console-consumer.sh --bootstrap-server kafka-node2:9092 --topic test-ken-io --from-beginning

然后均能收到消息
test by ken.io

這是因?yàn)檫@兩個(gè)消費(fèi)消息的命令是建立了兩個(gè)不同的Consumer
如果我們啟動(dòng)Consumer指定Consumer Group Id就可以作為一個(gè)消費(fèi)組協(xié)同工,1個(gè)消息同時(shí)只會(huì)被一個(gè)Consumer消費(fèi)到

bin/kafka-console-consumer.sh --bootstrap-server kafka-node3:9092 --topic test-ken-io --from-beginning --group testgroup_ken

bin/kafka-console-consumer.sh --bootstrap-server kafka-node2:9092 --topic test-ken-io --from-beginning --group testgroup_ken

參數(shù)

Kafka常用Broker配置說(shuō)明:

配置項(xiàng)默認(rèn)值/示例值說(shuō)明
broker.id0Broker唯一標(biāo)識(shí)
listenersPLAINTEXT://192.168.88.53:9092監(jiān)聽(tīng)信息,PLAINTEXT表示明文傳輸
log.dirskafka/logskafka數(shù)據(jù)存放地址,可以填寫多個(gè)。用”,”間隔
message.max.bytesmessage.max.bytes單個(gè)消息長(zhǎng)度限制,單位是字節(jié)
num.partitions1默認(rèn)分區(qū)數(shù)
log.flush.interval.messagesLong.MaxValue在數(shù)據(jù)被寫入到硬盤和消費(fèi)者可用前最大累積的消息的數(shù)量
log.flush.interval.msLong.MaxValue在數(shù)據(jù)被寫入到硬盤前的最大時(shí)間
log.flush.scheduler.interval.msLong.MaxValue檢查數(shù)據(jù)是否要寫入到硬盤的時(shí)間間隔。
log.retention.hours24控制一個(gè)log保留時(shí)間,單位:小時(shí)
zookeeper.connect192.168.88.21:2181ZooKeeper服務(wù)器地址,多臺(tái)用”,”間隔

看完上述內(nèi)容是否對(duì)您有幫助呢?如果還想對(duì)相關(guān)知識(shí)有進(jìn)一步的了解或閱讀更多相關(guān)文章,請(qǐng)關(guān)注億速云行業(yè)資訊頻道,感謝您對(duì)億速云的支持。

向AI問(wèn)一下細(xì)節(jié)

免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。

AI