Kafka基本使用

官網(wǎng)地址 ?http://kafka.apache.org/ ? 一切應(yīng)以官網(wǎng)文檔為準(zhǔn)。

?

安裝

download里下載要安裝的版本。或者直接wget該網(wǎng)址。如wget http://mirrors.cnnic.cn/apache/kafka/0.8.2.1/kafka_2.11-0.8.2.1.tgz

quickstart里有安裝方法：

tar -zxvf kafka_2.11-0.8.2.1.tgz

config/server.properties下有各種各種配置

先要保證zookeeper以啟動(dòng)，然后啟動(dòng)kafka ? ?bin/kafka-server-start.sh?config/server.properties?

?

?

測(cè)試

可以先測(cè)試一下：這些命令都以官網(wǎng)最新的為準(zhǔn)： ?建topic（指定zookeeper factor?partitions ?topic名） ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? 看topic （指定zookeeper） ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?bin/kafka-topics.sh --list --zookeeper ?localhost:2181

　　　　　　　　　　　　　　　　　　　　　　　啟動(dòng)consumer ?（指定zookeeper ?topic ? from開始） 啟動(dòng)后不要關(guān) ? ? ? ? ??bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

　　　　　　　　　　　　　　　　　　　　　　　再啟一個(gè)窗口 ?啟動(dòng)producer ?（指定broker topic） 輸入內(nèi)容回車，去consumer觀察 ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ??bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

?

報(bào)錯(cuò)

注意版本問題，如果javaApi producer版本高，想在客戶端consumer啟動(dòng)低版本驗(yàn)證，會(huì)不停的報(bào)錯(cuò)：

Closing socket connection to/127,0,0,1.(kafka.network.Processor) ?無法識(shí)別客戶端消息。

由于需要定時(shí)啟動(dòng)Kafka?consumer拉取數(shù)據(jù)，第一次啟動(dòng)后，沒有關(guān)掉線程。

但Kafka?consumer是非線程安全的，第二次消費(fèi)數(shù)據(jù)時(shí)會(huì)報(bào)錯(cuò)：

java.util.ConcurrentModificationException:?KafkaConsumer?is?not?safe?for?multi-threaded?access ?

轉(zhuǎn)載于:https://www.cnblogs.com/tree1123/p/6760688.html

總結(jié)

以上是生活随笔為你收集整理的Kafka基本使用的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。