2. 程式人生 > >Apache Kafka程式設計入門指南：設定分割槽數和複製因子

Apache Kafka程式設計入門指南：設定分割槽數和複製因子

阿新 • • 發佈：2019-02-13

我們學習瞭如何編寫簡單的Kafka Producer程式。在那個例子中，在如果需要傳送的topic不存在，Producer將會建立它。我們都知道（假設你知道），每個topic都是有分割槽數和複製因子的，但是我們無法通過Producer相關的API設定分割槽數和複製因子的，因為Producer相關API建立topic的是通過讀取server.properties檔案中的num.partitions和default.replication.factor的。那麼是否就意味著咱們無法在程式裡面定義topic的分割槽數和複製因子呢？

我們可以通過Kafka提供的AdminUtils.createTopic

函式來建立topic，它的函式原型如下：

def createTopic(zkClient: ZkClient, 
      topic: String,
      partitions: Int,   
      replicationFactor: Int,  
      topicConfig: Properties = new Properties)
這個函式是沒有返回值的。從上面的引數列表我們可以看出，partitions和replicationFactor引數就是上面說到的分割槽數和複製因子，所以我們可以通過這個引數來建立topic。在使用createTopic函式之前，我們需要建立zkClient物件，它裡面封裝了操作Zookeeper的相關API。而這個API不是Kafka內建的，所以我們需要先引入這個依賴：

<dependency>
      <groupId>com.101tec</groupId>
      <artifactId>zkclient</artifactId>
      <version>0.3</version>
</dependency>
然後我們就可以建立ZkClient物件了： 
val zk = "node1:2181"
val sessionTimeoutMs = 10000
val connectionTimeoutMs = 10000
val zkClient = new ZkClient(zk, sessionTimeoutMs, connectionTimeoutMs, ZKStringSerializer)
需要特別主要的是，我們必須指定ZKStringSerializer物件，否則執行完程式碼之後，你可以看到zookeeper裡面已經建立了相關topic，而且你list的時候也可以看到你建立的topic，但是當你往這個topic裡面傳送訊息的是，你會得到以下的異常：
[2016-02-05 16:45:52,335] ERROR Failed to collate messages by topic, partition due to: Failed to fetch topic metadata for topic: test-topic(kafka.producer.async.DefaultEventHandler)
[2016-02-05 16:45:52,441] WARN Error while fetching metadata [{TopicMetadata for topic iteblog ->
No partition metadata for topic flight due to kafka.common.LeaderNotAvailableException}] for topic [flight]: class kafka.common.LeaderNotAvailableException  (kafka.producer.BrokerPartitionInfo)
[2016-02-05 16:45:52,441] ERROR Failed to send requests for topics flight with correlation ids in [41,48] (kafka.producer.async.DefaultEventHandler)
[2016-02-05 16:45:52,441] ERROR Error in handling batch of 1 events (kafka.producer.async.ProducerSendThread)
kafka.common.FailedToSendMessageException: Failed to send messages after 3 tries.
    at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:90)
    at kafka.producer.async.ProducerSendThread.tryToHandle(ProducerSendThread.scala:105)
    at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:88)
    at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:68)
    at scala.collection.immutable.Stream.foreach(Stream.scala:547)
    at kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:67)
    at kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45)

找到topic的Leader，所以無法傳送訊息。這是因為如果你不使用ZKStringSerializer物件，那麼只會在Zookeeper裡面建立topic的相關資訊，但是kafka並沒有建立這個主題！現在我們就可以使用AdminUtils.createTopic參見topic了：

val topic = "test-topic"
val replicationFactor = 1
val numPartitions = 2
AdminUtils.createTopic(zkClient, topic, numPartitions, replicationFactor)
如果topic存在，那麼程式將會報錯： 
Exception in thread "main" kafka.common.TopicExistsException: Topic "test-topic" already exists.
    at kafka.admin.AdminUtils$.createOrUpdateTopicPartitionAssignmentPathInZK(AdminUtils.scala:187)
    at kafka.admin.AdminUtils$.createTopic(AdminUtils.scala:172)
    at com.iteblog.kafka.IteblogProducerV3$.main(IteblogProducerV3.scala:46)
    at com.iteblog.kafka.IteblogProducerV3.main(IteblogProducerV3.scala)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
    at java.lang.reflect.Method.invoke(Method.java:597)
    at com.intellij.rt.execution.application.AppMain.main(AppMain.java:144)

否則什麼都沒輸出則代表你的topic建立成功了！完整程式碼片段如下： 
package com.practice.kafka

import kafka.admin.AdminUtils
import kafka.utils.ZKStringSerializer
import org.I0Itec.zkclient.ZkClient
 
object CreateTopic {

  def main(args: Array[String]) {
    val zk = "node1:2181"
    val sessionTimeoutMs = 10000
    val connectionTimeoutMs = 10000
    val zkClient = new ZkClient(zk, sessionTimeoutMs, connectionTimeoutMs, ZKStringSerializer)
    val topic = "test-topic"
    val replicationFactor = 1
    val numPartitions = 2
    AdminUtils.createTopic(zkClient, topic, numPartitions, replicationFactor)
  }
}
除了使用上面AdminUtils.createTopic在建立主題的時候設定複製因子和分割槽數，我們還可以使用kafka.admin.TopicCommand來實現同樣的功能，如下： 
val arguments = Array("--create", "--zookeeper", zk, "--replication-factor", "2", "--partition", "2", "--topic", "test-topic")
TopicCommand.main(arguments)

相關推薦

Apache Kafka程式設計入門指南：設定分割槽數和複製因子

我們學習瞭如何編寫簡單的Kafka Producer程式。在那個例子中，在如果需要傳送的topic不存在，Producer將會建立它。我們都知道（假設你知道），每個topic都是有分割槽數和複製因子的，但是我們無法通過Producer相關的API設定分割槽數和複製因子的，

Apache Zeppelin使用入門指南：程式設計

　　編譯和啟動完相關的程序之後，我們就可以來使用了。我們進入到http://www.iteblog.com:8080頁面，我們可以在頁面上直接操作Zeppelin，依次選擇Notebook->Create new note，然後會彈出一個對話方塊，我

【python】入門指南：控制語句

pan else pre 循環 clas python continue break for 條件控制 if，if-else，if-elseif-else #!/bin/python a = ‘test‘ if a == ‘test‘: print

Shell程式設計入門四：函式

使用者可以用shell定義函式，然後子啊shell指令碼中隨便呼叫。shell中函式的定義格式如下： [ function ] funname [()] { action; [return int;] } 可以帶 function fun() 定義，也可以

Shell程式設計入門三：流程控制

和Java、PHP等語言不一樣，sh的流程控制不可為空，如： public class Test{ public static void main(String[] args){ int a = 10; if(a > 5){

Shell程式設計入門二：echo命令

Shell的 echo 命令和PHP的echo指令類似，都是用於輸出值。我們可以使用 echo 實現更復雜的輸出格式控制。 顯示普通字串 echo "Hello World" echo Hello World 這兩句輸出結果均為：Hello World 顯示轉義字元

Shell程式設計入門一：傳遞引數

我們可以在執行 Shell 指令碼時，向指令碼傳遞引數，指令碼內獲取引數的格式為：$n 。n 代表一個數字，1 為執行指令碼的第一個引數，2 為執行指令碼的第二個引數，以此類推…… 例項 我們向指令碼傳遞三個引數，並分別輸出。其中 $0 表示執行的檔名: #! /bin/ba

【python】入門指南：常用資料結構

Python內建了三種高階資料結構：list，tuple，dict list：陣列，相同型別的元素組成的陣列 tuple：元組，相同型別的元素組成的陣列，但是這裡有限定條件（長度是固定的，並且值也是固定的，不能被改變） dict：字典，k-v結構的   list陣列 1，初始化和遍歷li

「技術乾貨」python程式設計入門須知：python運算子

  學習一定是腳踏實地一步一個腳印的，程式設計也是一樣的。想要學好Python，首先來了解一遍Python程式設計入門知識之Python運算子吧。 運算子的知識主要是分為以下幾點： 1、算術運算子 2、比較(即關係)運算子 3、賦值運算子 4、邏輯運算子

Maven入門指南：倉庫

轉載https://blog.csdn.net/a258831020/article/details/49965803 1 . 倉庫簡介 沒有 Maven 時，專案用到的 .jar 檔案通常需要拷貝到 /lib 目錄，專案多了，拷貝的檔案副本就多了，佔用磁碟空間，且難於管理。Maven 使用

腦殘式網路程式設計入門(六)：什麼是公網IP和內網IP？NAT轉換又是什麼鬼？

本文引用了“帥地”發表於公眾號苦逼的碼農的技術分享。 1、引言 搞網路通訊應用開發的程式設計師，可能會經常聽到外網IP（即網際網路IP地址）和內網IP（即區域網IP地址），但他們的區別是什麼？又有什麼關係呢？另外，內行都知道，提到外網IP和內網IP就不得不提NAT路由轉換這種東西，那這雙是什

腦殘式網路程式設計入門(五)：每天都在用的Ping命令，它到底是什麼？

本文引用了公眾號純潔的微笑作者奎哥的技術文章，感謝原作者的分享。 1、前言 老於網路程式設計熟手來說，在測試和部署網路通訊應用（比如IM聊天、實時音視訊等）時，如果發現網路連線超時，第一時間想到的就是使用Ping命令Ping一下伺服器看看通不通。甚至在有些情況下通過圖

Spark DataFrames入門指南：建立和操作DataFrame

一、從csv檔案建立DataFrame 　　本文將介紹如何從csv檔案建立DataFrame。 如何做？ 　　從csv檔案建立DataFrame主要包括以下幾步驟： 　　1、在build.sbt檔案裡面新增spark-csv支援庫； 　　2、建立Sp

Kubernetes入門指南：部署一個Node.js Docker應用 — SitePoint

容器技術已經存在了很多年，Docker使它真正成為主流。因為Docker提供了一套簡單易用的介面來使用容器，現在已有許多的公司和開發者使用容器來裝載他們的應用。 然而，對於任何有一定規模的應用程式，你一定不會只要部署“一個容器”，而是部署在大量主機上的許多

張高興的 .NET Core IoT 入門指南：（三）使用 I2C 進行通訊

什麼是 I2C 匯流排 I2C 匯流排（Inter-Integrated Circuit Bus）是裝置與裝置間通訊方式的一種。它是一種序列通訊匯流排，由飛利浦公司在1980年代為了讓主機板、嵌入式系統或手機用以連線低速周邊裝置而發展[1]。I2C 匯流排包含兩根訊號線，一根為訊號線 SDA ，另一根為時鐘

張高興的 .NET Core IoT 入門指南：（四）使用 SPI 進行通訊

什麼是 SPI 和上一篇文章的 I2C 匯流排一樣，SPI（Serial Peripheral Interface，序列外設介面）也是裝置與裝置間通訊方式的一種。SPI 是一種全雙工（資料可以兩個方向同時傳輸）的序列通訊匯流排，由摩托羅拉於上個世紀 80 年代開發[1]，用於短距離裝置之間的通訊。SPI 包

張高興的 .NET Core IoT 入門指南：（五）串列埠通訊入門

在開始之前，首先要說明的是串列埠通訊所用到的 SerialPort 類並不包含在 System.Device.Gpio NuGet 包中，而是在 System.IO.Ports NuGet 包中。之所以在這裡介紹串列埠通訊，是因為在嵌入式中串列埠通訊是與其他裝置進行互動的一種重要方式，而且在某些沒有螢幕的裝

張高興的 .NET Core IoT 入門指南：（五）PWM 訊號輸出

什麼是 PWM 在解釋 PWM 之前首先來了解一下電路中訊號的概念，其中包括模擬訊號和數字訊號。模擬訊號是一種連續的訊號，與連續函式類似，在圖形上表現為一條不間斷的連續曲線。數字訊號為只能取有限個數值的訊號，比如計算機中的高電平（1）和低電平（0）。 PWM（Pulse Width Modulation）即

Spark Streaming 程式設計入門指南

Spark Streaming 是核心Spark API的擴充套件，可實現實時資料流的可伸縮，高吞吐量，容錯流處理。可以從許多資料來源（例如Kafka，Flume，Kinesis或TCP sockets）中提取資料，並且可以使用複雜的演算法處理資料，這些演算法用高階函式表示，如map、reduce、join和

Jmeter Web 性能測試入門 (五)：Jmeter 參數化 Request

user clas 測試入門 pre 參數化 需要 data .get 9.png 用來參數化的常用方法： 添加配置元件：用戶定義的變量 使用函數助手 添加配置元件：CSV Data Set Config 添加前置處理器：BeanShell PreProcessor 添