轉(zhuǎn)自： Java程序創(chuàng)建Kafka Topic，以及數(shù)據(jù)生產(chǎn)消費，常用的命令_Zyy_z_的博客-CSDN博客_java kafka創(chuàng)建topicKafka簡介： Kafka是一個分布式發(fā)布——訂閱消息傳遞系統(tǒng)。Kafka快速、可擴展且耐用。它保留主題中的消息源。生產(chǎn)者將數(shù)據(jù)寫入主題，消費者從主題中讀取數(shù)據(jù)。Kafka的特點： 1. 同時為分布和訂閱提供高吞吐量。據(jù)了解，Kafka每秒可以生產(chǎn)約25萬條消息（50MB），每秒處理55萬條消息...https://blog.csdn.net/Zyy_z_/article/details/101680138

---

【1】Kafka簡介

Kafka是一個分布式發(fā)布——訂閱消息傳遞系統(tǒng)。Kafka快速、可擴展且耐用。它保留主題中的消息源。生產(chǎn)者將數(shù)據(jù)寫入主題，消費者從主題中讀取數(shù)據(jù)。

1）Kafka的特點：

• 1. 同時為分布和訂閱提供高吞吐量。 據(jù)了解，Kafka每秒可以生產(chǎn)約25萬條消息（50MB），每秒處理55萬條消息（110MB）這里說條數(shù)，可能不上特別準(zhǔn)確，因為消息的大小可能不一致；
• 2. 可進行持久化操作，將消息持久化到到磁盤，以日志的形式存儲，因此可用于批量消費，例如ETL，以及實時應(yīng)用程序。 通過將數(shù)據(jù)持久化到硬盤以及replication防止數(shù)據(jù)丟失。
• 3. 分布式系統(tǒng)，易于向外拓展。所有的Producer、broker和consumer都會有多個，均為分布式。無需停機即可拓展 機器。
• 4. 消息被處理的狀態(tài)是在consumer端維護，而不是由server端維護，當(dāng)失敗時能自動平衡。

2）Kafka名詞解釋：

producer：消息的生成者

consumer：消息的消費者

topic：你把它理解為標(biāo)簽

broker：Kafka處理資源的消息源(feeds of messages)的不同分類

3）Kafka常用命令:

創(chuàng)建主題（4個分區(qū)，2個副本):? kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 2 --partitions 4 --topic test

查詢所有Topic：kafka-topics.sh --zookeeper localhost:2181 --list

查看指定得Topic：kafka-topics.sh --zookeeper localhost:2181 --describe --topic t_cdr

刪除Topic：kafka-run-class.sh kafka.admin.DeleteTopicCommand --zookeeper localhost:2181 --topic t_cdr

生產(chǎn)者 ：kafka-console-producer.sh --broker-list localhost:9092 --topic test

消費者 : kafka-console-consumer.sh --zookeeper localhost:2181 --topic test

新生產(chǎn)者（支持0.9版本+）:? kafka-console-producer.sh --broker-list localhost:9092 --topic test --producer.config config/producer.properties

新消費者（支持0.9版本+）:? kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --new-consumer --from-beginning --consumer.config config/consumer.properties

---

【2】kafka java api

【2.1】Java程序操作創(chuàng)建Topic：? ?

Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("acks", "all"); props.put("retries", 0); props.put("batch.size", 16384); props.put("linger.ms", 1); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); AdminClient create = KafkaAdminClient.create(props);//創(chuàng)建Topic create.createTopics(Lists.newArrayList(new NewTopic("Topic名稱")，1，（short）1));//一個分區(qū) create.close();//關(guān)閉

其他創(chuàng)建Topic得方式Java API：

https://blog.csdn.net/meng984611383/article/details/80500761https://blog.csdn.net/meng984611383/article/details/80500761

---

【2.2】Kafka生產(chǎn)數(shù)據(jù):

Producer<String, String> producer = new KafkaProducer<>(props); for(int i = 0; i < 100; i++) //生產(chǎn)數(shù)據(jù)producer.send(new ProducerRecord<String, String>("Topic名稱", Integer.toString(i), Integer.toString(i))); producer.close(); //關(guān)閉

---

【2.3】消費數(shù)據(jù)

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); consumer.subscribe(Arrays.asList("foo", "bar")); while (true) {ConsumerRecords<String, String> records = consumer.poll(100);for (ConsumerRecord<String, String> record : records)System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());

生產(chǎn)者的緩沖空間池保留尚未發(fā)送到服務(wù)器的消息，后臺I/O線程負責(zé)將這些消息轉(zhuǎn)換成請求發(fā)送到集群。如果使用后不關(guān)閉生產(chǎn)者，則會泄露這些資源。

send()方法是異步的，添加消息到緩沖區(qū)等待發(fā)送，并立即返回。生產(chǎn)者將單個的消息批量在一起發(fā)送來提高效率。

ack是判別請求是否為完整的條件（就是是判斷是不是成功發(fā)送了）。我們指定了“all”將會阻塞消息，這種設(shè)置性能最低，但是是最可靠的。

retries，如果請求失敗，生產(chǎn)者會自動重試，我們指定是0次，如果啟用重試，則會有重復(fù)消息的可能性。

producer(生產(chǎn)者)緩存每個分區(qū)未發(fā)送的消息。緩存的大小是通過 batch.size 配置指定的。值較大的話將會產(chǎn)生更大的批。并需要更多的內(nèi)存（因為每個“活躍”的分區(qū)都有1個緩沖區(qū)）。

默認(rèn)緩沖可立即發(fā)送，即便緩沖空間還沒有滿，但是，如果你想減少請求的數(shù)量，可以設(shè)置linger.ms大于0。這將指示生產(chǎn)者發(fā)送請求之前等待一段時間，希望更多的消息填補到未滿的批中。這類似于TCP的算法，例如上面的代碼段，可能100條消息在一個請求發(fā)送，因為我們設(shè)置了linger(逗留)時間為1毫秒，然后，如果我們沒有填滿緩沖區(qū)，這個設(shè)置將增加1毫秒的延遲請求以等待更多的消息。需要注意的是，在高負載下，相近的時間一般也會組成批，即使是 linger.ms=0。在不處于高負載的情況下，如果設(shè)置比0大，以少量的延遲代價換取更少的，更有效的請求。

buffer.memory 控制生產(chǎn)者可用的緩存總量，如果消息發(fā)送速度比其傳輸?shù)椒?wù)器的快，將會耗盡這個緩存空間。當(dāng)緩存空間耗盡，其他發(fā)送調(diào)用將被阻塞，阻塞時間的閾值通過max.block.ms設(shè)定，之后它將拋出一個TimeoutException。

key.serializer和value.serializer示例，將用戶提供的key和value對象ProducerRecord轉(zhuǎn)換成字節(jié)，你可以使用附帶的ByteArraySerializaer或StringSerializer處理簡單的string或byte類型。

?
?

總結(jié)

以上是生活随笔為你收集整理的Java程序创建Kafka Topic，以及数据生产消费，常用的命令的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

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