Kafka簡單的生產者例子


在這一節中將建立一個使用Java用戶端發布和使用訊息的應用程式。 Kafka生產者用戶端由以下API組成。

KafkaProducer API

下面來了解Kafka生產者API。 KafkaProducer API的核心部分是KafkaProducer類。 KafkaProducer類提供了一個選項,用於將Kafka代理的建構函式與以下方法連線起來。

  • KafkaProducer類提供send()方法來非同步傳送訊息到主題。 send()的簽名如下 -
    producer.send(new ProducerRecord<byte[],byte[]>(topic, partition, key1, value1) , callback);
    
  • ProducerRecord - 生產者管理等待傳送的記錄緩衝區。
  • Callback - 伺服器確認記錄時執行的使用者提供的回撥函式(null表示無回撥)。
  • KafkaProducer類提供了一個flush方法來確保所有先前傳送的訊息已經實際完成。 flush方法的語法如下 -
    public void flush()
    
  • KafkaProducer類提供了partitionFor方法,該方法有助於獲取給定主題的分割區後設資料。 這可以用於自定義分割區。 這種方法的簽名如下 -
    public Map metrics()
    
    它返回生產者維護的內部度量圖。
  • public void close() - KafkaProducer類提供close方法,阻塞直到完成所有先前傳送的請求。

生產者API

Producer API的核心部分是Producer類。 Producer類提供了一個選項,通過以下方法在其建構函式中連線Kafka代理。

Producer類

Producer類提供send方法,使用以下簽名將訊息傳送到單個或多個主題。

public void send(KeyedMessaget<k,v> message) 
- sends the data to a single topic,par-titioned by key using either sync or async producer.
public void send(List<KeyedMessage<k,v>>messages)
- sends data to multiple topics.
Properties prop = new Properties();
prop.put(producer.type,」async」)
ProducerConfig config = new ProducerConfig(prop);

有兩種型別的生產者 - 同步和非同步。

同樣的API組態也適用於Sync生產者。 它們之間的區別是同步生產者直接傳送訊息,但在後台傳送訊息。 當想要更高的吞吐量時,首選非同步生產者。 在0.8之前的版本中,非同步生產者沒有回撥send()來註冊錯誤處理程式。非同步生產者僅在當前版本0.9中可用。

public void close()

Producer類提供close()方法來關閉所有Kafka 經紀人的生產者池連線。

組態設定

Producer API的主要組態設定在下表中列出以便更好地理解 -

編號 組態設定 描述
1 client.id 確定生產者應用
2 producer.type 是同步還是非同步?
3 acks acks組態控制生產者請求下的標準被認為是完整的。
4 retries 如果生產者請求失敗,則自動重試具有特定的值。
5 bootstrap.servers 引導經紀人(brokers)列表。
6 linger.ms 如果想減少請求的數量,可以將linger.ms設定為大於某個值的值。
7 key.serializer 序列化器介面的鍵。
8 value.serializer 序列化器介面的值。
9 batch.size 緩衝區大小。
10 buffer.memory 控制生產者可用於緩衝的總記憶體量。

ProducerRecord API

ProducerRecord是傳送給Kafka的cluster.ProducerRecord類的鍵/值對,用於使用以下簽名建立包含分割區,鍵和值對的記錄。

public ProducerRecord (string topic, int partition, k key, v value)
  • topic ? 使用者定義的主題名稱將要追加到記錄中。
  • partition ? 分割區計數
  • key ? 記錄中將包含的關鍵字。
  • value ? 記錄內容
public ProducerRecord (string topic, k key, v value)

ProducerRecord類建構函式用於建立一個記錄,其中包含鍵值對,但沒有分割區。

  • topic ? 建立一個分配記錄的主題。
  • partition ? 分割區計數
  • key ? 記錄中將包含的關鍵字。
  • value ? 記錄內容
public ProducerRecord (string topic, v value)

ProducerRecord類建立沒有分割區和鍵的記錄。

  • topic ? 建立一個分配記錄的主題。
  • value ? 記錄內容

下表中列出了ProducerRecord類的方法 -

編號 方法 描述
1 public string topic() 將附加到記錄的主題。
2 public K key() 將包含在記錄中的鍵。 如果沒有這樣的鍵,返回null
3 public V value() 記錄的內容。
4 partition() 記錄的分割區計數

SimpleProducer應用程式

在建立應用程式之前,首先啟動ZooKeeper和Kafka代理,然後使用create topic命令在Kafka代理中建立自己的主題。 之後,建立一個SimpleProducer.java的Java類並輸入以下編碼。

//import util.properties packages
import java.util.Properties;

//import simple producer packages
import org.apache.kafka.clients.producer.Producer;

//import KafkaProducer packages
import org.apache.kafka.clients.producer.KafkaProducer;

//import ProducerRecord packages
import org.apache.kafka.clients.producer.ProducerRecord;

//Create java class named 「SimpleProducer」
public class SimpleProducer {

   public static void main(String[] args) throws Exception{

      // Check arguments length value
      if(args.length == 0){
         System.out.println("Enter topic name」);
         return;
      }

      //Assign topicName to string variable
      String topicName = args[0].toString();

      // create instance for properties to access producer configs   
      Properties props = new Properties();

      //Assign localhost id
      props.put("bootstrap.servers", 「localhost:9092");

      //Set acknowledgements for producer requests.      
      props.put("acks", 「all");

      //If the request fails, the producer can automatically retry,
      props.put("retries", 0);

      //Specify buffer size in config
      props.put("batch.size", 16384);

      //Reduce the no of requests less than 0   
      props.put("linger.ms", 1);

      //The buffer.memory controls the total amount of memory available to the producer for buffering.   
      props.put("buffer.memory", 33554432);

      props.put("key.serializer", 
         "org.apache.kafka.common.serializa-tion.StringSerializer");

      props.put("value.serializer", 
         "org.apache.kafka.common.serializa-tion.StringSerializer");

      Producer<String, String> producer = new KafkaProducer
         <String, String>(props);

      for(int i = 0; i < 10; i++)
         producer.send(new ProducerRecord<String, String>(topicName, 
            Integer.toString(i), Integer.toString(i)));
               System.out.println(「Message sent successfully」);
               producer.close();
   }
}

編譯 - 可以使用以下命令編譯應用程式 -

$ javac -cp 「/path/to/kafka/kafka_2.11-0.9.0.0/lib/*」 *.java

執行 - 可以使用以下命令執行應用程式。

java -cp 「/path/to/kafka/kafka_2.11-0.9.0.0/lib/*」:. SimpleProducer <topic-name>

執行上面範例程式碼,得到以下結果 -

Message sent successfully
To check the above output open new terminal and type Consumer CLI command to receive messages.
>> bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic <topic-name> —from-beginning
1
2
3
4
5
6
7
8
9
10

簡單的消費者範例

截至目前,已經建立了一個生產者並向Kafka叢集傳送訊息。 現在,建立一個消費者來使用來自Kafka叢集的訊息。 KafkaConsumer API用於消費來自Kafka叢集的訊息。 下面定義了KafkaConsumer類別建構函式。

public KafkaConsumer(java.util.Map<java.lang.String,java.lang.Object> configs)
  • configs - 返回消費者組態的對映。

KafkaConsumer類具有下表中列出的方法。

編號 方法 描述
1 public java.util.Set<TopicPar-tition> assignment() 獲取消費者當前分配的一組分割區。
2 public string subscription() 訂閱給定的主題列表以獲取動態簽署的分割區。
3 public void sub-scribe(java.util.List<java.lang.String> topics, ConsumerRe-balanceListener listener) 訂閱給定的主題列表以獲取動態簽署的分割區。
4 public void unsubscribe() 取消訂閱給定分割區列表中的主題。
5 public void sub-scribe(java.util.List<java.lang.String> topics) 訂閱給定的主題列表以獲取動態簽署的分割區。 如果給定的主題列表為空,則將其視為與unsubscribe()相同。
6 public void sub-scribe(java.util.regex.Pattern pattern, ConsumerRebalanceLis-tener listener) 引數(pattern)以正規表示式的格式參照訂閱模式,而引數listener從訂閱模式中獲取通知。
7 public void as-sign(java.util.List<TopicParti-tion> partitions) 手動將分割區列表分配給客戶。
8 poll() 獲取使用其中一個訂閱/分配API指定的主題或分割區的資料。 如果在輪詢資料之前沒有訂閱主題,這將返回錯誤。
9 public void commitSync() 提交最後一次由poll()返回的主題和分割區的所有sub-scribed列表的偏移量。 同樣的操作應用於commitAsyn()
10 public void seek(TopicPartition partition, long offset) 獲取消費者將在下一個poll()方法中使用的當前偏移值。
11 public void resume() 恢復暫停的分割區。
12 public void wakeup() 喚醒消費者。

ConsumerRecord API

ConsumerRecord API用於接收來自Kafka叢集的記錄。 該API由主題名稱,分割區編號以及指向Kafka分割區中記錄的偏移量組成。 ConsumerRecord類用於建立具有特定主題名稱,分割區計數和<key,value>對的消費者記錄。 它有以下簽名。

public ConsumerRecord(string topic,int partition, long offset,K key, V value)

引數 -

  • topic - 從Kafka叢集收到的消費者記錄的主題名稱。
  • partition - 主題的分割區。
  • key - 記錄的關鍵字,如果不存在關鍵字,則將返回null
  • value - 記錄內容。

ConsumerRecords API

ConsumerRecords API充當ConsumerRecord的容器。 此API用於保留特定主題的每個分割區的ConsumerRecord列表。 它的建構函式定義如下。

public ConsumerRecords(java.util.Map<TopicPartition,java.util.List
<Consumer-Record>K,V>>> records)

引數

  • TopicPartition - 返回特定主題的分割區對映。
  • records - ConsumerRecord的返回列表。

ConsumerRecords類中定義了以下方法。

編號 方法 描述
1 public int count() 所有主題的記錄數。
2 public Set partitions() 該記錄集中包含資料的分割區集(如果沒有資料返回,則該集為空)。
3 public Iterator iterator() 疊代器能夠遍歷集合,獲取或重新移動元素。
4 public List records() 獲取給定分割區的記錄列表。

組態設定

以下列出了Consumer用戶端API主組態設定的組態設定 -

編號 設定 描述
1 bootstrap.servers 經紀人的引導列表。
2 group.id 將個人消費者分配給組。
3 enable.auto.commit 如果值為true,則為偏移啟用自動提交,否則不提交。
4 auto.commit.interval.ms 返回更新的消耗偏移量被寫入ZooKeeper的頻率。
5 session.timeout.ms 指示在放棄並繼續使用訊息之前,Kafka將等待多少毫秒以等待ZooKeeper響應請求(讀取或寫入)。

SimpleConsumer應用程式

生產者申請步驟在這裡保持不變。 首先,啟動ZooKeeper和Kafka經紀人。 然後使用SimpleConsumer.java的Java類建立一個SimpleConsumer應用程式並鍵入以下程式碼。

import java.util.Properties;
import java.util.Arrays;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;

public class SimpleConsumer {
   public static void main(String[] args) throws Exception {
      if(args.length == 0){
         System.out.println("Enter topic name");
         return;
      }
      //Kafka consumer configuration settings
      String topicName = args[0].toString();
      Properties props = new Properties();

      props.put("bootstrap.servers", "localhost:9092");
      props.put("group.id", "test");
      props.put("enable.auto.commit", "true");
      props.put("auto.commit.interval.ms", "1000");
      props.put("session.timeout.ms", "30000");
      props.put("key.deserializer", 
         "org.apache.kafka.common.serializa-tion.StringDeserializer");
      props.put("value.deserializer", 
         "org.apache.kafka.common.serializa-tion.StringDeserializer");
      KafkaConsumer<String, String> consumer = new KafkaConsumer
         <String, String>(props);

      //Kafka Consumer subscribes list of topics here.
      consumer.subscribe(Arrays.asList(topicName))

      //print the topic name
      System.out.println("Subscribed to topic " + topicName);
      int i = 0;

      while (true) {
         ConsumerRecords<String, String> records = con-sumer.poll(100);
         for (ConsumerRecord<String, String> record : records)

         // print the offset,key and value for the consumer records.
         System.out.printf("offset = %d, key = %s, value = %s\n", 
            record.offset(), record.key(), record.value());
      }
   }
}

編譯 - 可以使用以下命令編譯應用程式。

javac -cp 「/path/to/kafka/kafka_2.11-0.9.0.0/lib/*」 *.java

執行 - 可以使用以下命令執行應用程式 -

java -cp 「/path/to/kafka/kafka_2.11-0.9.0.0/lib/*」:. SimpleConsumer <topic-name>

輸入 - 開啟生產者CLI並行送一些訊息給主題。輸入如'Hello Consumer'。得到以類似以下結果 -

Subscribed to topic Hello-Kafka
offset = 3, key = null, value = Hello Consumer