Kafka利用Java實(shí)現(xiàn)數(shù)據(jù)的生產(chǎn)和消費(fèi)實(shí)例教程

更新時(shí)間：2018年01月28日 16:27:46 作者：虛無境

這篇文章主要給大家介紹了關(guān)于Kafka利用Java實(shí)現(xiàn)數(shù)據(jù)的生產(chǎn)和消費(fèi)的相關(guān)資料，文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧。

前言

在上一篇中講述如何搭建kafka集群，本篇?jiǎng)t講述如何簡單的使用 kafka 。不過在使用kafka的時(shí)候，還是應(yīng)該簡單的了解下kafka。

Kafka的介紹

Kafka是一種高吞吐量的分布式發(fā)布訂閱消息系統(tǒng)，它可以處理消費(fèi)者規(guī)模的網(wǎng)站中的所有動(dòng)作流數(shù)據(jù)。

Kafka 有如下特性：

以時(shí)間復(fù)雜度為O(1)的方式提供消息持久化能力，即使對(duì)TB級(jí)以上數(shù)據(jù)也能保證常數(shù)時(shí)間復(fù)雜度的訪問性能。
高吞吐率。即使在非常廉價(jià)的商用機(jī)器上也能做到單機(jī)支持每秒100K條以上消息的傳輸。
支持Kafka Server間的消息分區(qū)，及分布式消費(fèi)，同時(shí)保證每個(gè)Partition內(nèi)的消息順序傳輸。
同時(shí)支持離線數(shù)據(jù)處理和實(shí)時(shí)數(shù)據(jù)處理。
Scale out：支持在線水平擴(kuò)展。

kafka的術(shù)語

Broker：Kafka集群包含一個(gè)或多個(gè)服務(wù)器，這種服務(wù)器被稱為broker。
Topic：每條發(fā)布到Kafka集群的消息都有一個(gè)類別，這個(gè)類別被稱為Topic。（物理上不同Topic的消息分開存儲(chǔ)，邏輯上一個(gè)Topic的消息雖然保存于一個(gè)或多個(gè)broker上但用戶只需指定消息的Topic即可生產(chǎn)或消費(fèi)數(shù)據(jù)而不必關(guān)心數(shù)據(jù)存于何處）
Partition：Partition是物理上的概念，每個(gè)Topic包含一個(gè)或多個(gè)Partition。
Producer：負(fù)責(zé)發(fā)布消息到Kafka broker。
Consumer：消息消費(fèi)者，向Kafka broker讀取消息的客戶端。
Consumer Group:每個(gè)Consumer屬于一個(gè)特定的Consumer Group（可為每個(gè)Consumer指定group name，若不指定group name則屬于默認(rèn)的group）。

kafka核心Api

kafka有四個(gè)核心API

應(yīng)用程序使用producer API發(fā)布消息到1個(gè)或多個(gè)topic中。
應(yīng)用程序使用consumer API來訂閱一個(gè)或多個(gè)topic,并處理產(chǎn)生的消息。
應(yīng)用程序使用streams API充當(dāng)一個(gè)流處理器,從1個(gè)或多個(gè)topic消費(fèi)輸入流,并產(chǎn)生一個(gè)輸出流到1個(gè)或多個(gè)topic,有效地將輸入流轉(zhuǎn)換到輸出流。
connector API允許構(gòu)建或運(yùn)行可重復(fù)使用的生產(chǎn)者或消費(fèi)者,將topic鏈接到現(xiàn)有的應(yīng)用程序或數(shù)據(jù)系統(tǒng)。

示例圖如下:

kafka 應(yīng)用場(chǎng)景

構(gòu)建可在系統(tǒng)或應(yīng)用程序之間可靠獲取數(shù)據(jù)的實(shí)時(shí)流數(shù)據(jù)管道。
構(gòu)建實(shí)時(shí)流應(yīng)用程序，可以轉(zhuǎn)換或響應(yīng)數(shù)據(jù)流。

以上介紹參考kafka官方文檔。

開發(fā)準(zhǔn)備

如果我們要開發(fā)一個(gè)kafka的程序，應(yīng)該做些什么呢？

首先，在搭建好kafka環(huán)境之后，我們要考慮的是我們是生產(chǎn)者還是消費(fèi)者，也就是消息的發(fā)送者還是接受者。
不過在本篇中，生產(chǎn)者和消費(fèi)者都會(huì)進(jìn)行開發(fā)和講解。

在大致的了解kafka之后，我們來開發(fā)第一個(gè)程序。

這里用的開發(fā)語言是Java，構(gòu)建工具M(jìn)aven。

Maven的依賴如下:

 <dependency>
  <groupId>org.apache.kafka</groupId>
   <artifactId>kafka_2.12</artifactId>
   <version>1.0.0</version>
   <scope>provided</scope> 
  </dependency>
  
  <dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>1.0.0</version>
  </dependency>
  
  <dependency>
   <groupId>org.apache.kafka</groupId>
   <artifactId>kafka-streams</artifactId>
   <version>1.0.0</version>
  </dependency>

Kafka Producer

在開發(fā)生產(chǎn)的時(shí)候，先簡單的介紹下kafka各種配置說明：

bootstrap.servers： kafka的地址。
acks:消息的確認(rèn)機(jī)制，默認(rèn)值是0。
acks=0：如果設(shè)置為0，生產(chǎn)者不會(huì)等待kafka的響應(yīng)。
acks=1：這個(gè)配置意味著kafka會(huì)把這條消息寫到本地日志文件中，但是不會(huì)等待集群中其他機(jī)器的成功響應(yīng)。
acks=all：這個(gè)配置意味著leader會(huì)等待所有的follower同步完成。這個(gè)確保消息不會(huì)丟失，除非kafka集群中所有機(jī)器掛掉。這是最強(qiáng)的可用性保證。
retries：配置為大于0的值的話，客戶端會(huì)在消息發(fā)送失敗時(shí)重新發(fā)送。
batch.size:當(dāng)多條消息需要發(fā)送到同一個(gè)分區(qū)時(shí)，生產(chǎn)者會(huì)嘗試合并網(wǎng)絡(luò)請(qǐng)求。這會(huì)提高client和生產(chǎn)者的效率。
key.serializer: 鍵序列化，默認(rèn)org.apache.kafka.common.serialization.StringDeserializer。
value.deserializer:值序列化，默認(rèn)org.apache.kafka.common.serialization.StringDeserializer。

...

還有更多配置，可以去查看官方文檔，這里就不在說明了。

那么我們kafka 的producer配置如下:

  Properties props = new Properties();
  props.put("bootstrap.servers", "master:9092,slave1:9092,slave2:9092");
  props.put("acks", "all");
  props.put("retries", 0);
  props.put("batch.size", 16384);
  props.put("key.serializer", StringSerializer.class.getName());
  props.put("value.serializer", StringSerializer.class.getName());
  KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);

kafka的配置添加之后，我們便開始生產(chǎn)數(shù)據(jù)，生產(chǎn)數(shù)據(jù)代碼只需如下就行:

producer.send(new ProducerRecord<String, String>(topic,key,value));

topic: 消息隊(duì)列的名稱，可以先行在kafka服務(wù)中進(jìn)行創(chuàng)建。如果kafka中并未創(chuàng)建該topic，那么便會(huì)自動(dòng)創(chuàng)建！
key:鍵值，也就是value對(duì)應(yīng)的值，和Map類似。
value:要發(fā)送的數(shù)據(jù)，數(shù)據(jù)格式為String類型的。

在寫好生產(chǎn)者程序之后，那我們先來生產(chǎn)吧！

我這里發(fā)送的消息為:

 String messageStr="你好，這是第"+messageNo+"條數(shù)據(jù)";

并且只發(fā)送1000條就退出，結(jié)果如下:

可以看到信息成功的打印了。

如果不想用程序進(jìn)行驗(yàn)證程序是否發(fā)送成功，以及消息發(fā)送的準(zhǔn)確性，可以在kafka服務(wù)器上使用命令查看。

Kafka Consumer

kafka消費(fèi)這塊應(yīng)該來說是重點(diǎn)，畢竟大部分的時(shí)候，我們主要使用的是將數(shù)據(jù)進(jìn)行消費(fèi)。

kafka消費(fèi)的配置如下:

bootstrap.servers： kafka的地址。
group.id：組名不同組名可以重復(fù)消費(fèi)。例如你先使用了組名A消費(fèi)了kafka的1000條數(shù)據(jù)，但是你還想再次進(jìn)行消費(fèi)這1000條數(shù)據(jù)，并且不想重新去產(chǎn)生，那么這里你只需要更改組名就可以重復(fù)消費(fèi)了。
enable.auto.commit：是否自動(dòng)提交，默認(rèn)為true。
auto.commit.interval.ms: 從poll(拉)的回話處理時(shí)長。
session.timeout.ms:超時(shí)時(shí)間。
max.poll.records:一次最大拉取的條數(shù)。
auto.offset.reset：消費(fèi)規(guī)則，默認(rèn)earliest 。
earliest: 當(dāng)各分區(qū)下有已提交的offset時(shí)，從提交的offset開始消費(fèi)；無提交的offset時(shí)，從頭開始消費(fèi) 。
latest: 當(dāng)各分區(qū)下有已提交的offset時(shí)，從提交的offset開始消費(fèi)；無提交的offset時(shí)，消費(fèi)新產(chǎn)生的該分區(qū)下的數(shù)據(jù) 。
none: topic各分區(qū)都存在已提交的offset時(shí)，從offset后開始消費(fèi)；只要有一個(gè)分區(qū)不存在已提交的offset，則拋出異常。
key.serializer: 鍵序列化，默認(rèn)org.apache.kafka.common.serialization.StringDeserializer。
value.deserializer:值序列化，默認(rèn)org.apache.kafka.common.serialization.StringDeserializer。

那么我們kafka 的consumer配置如下:

 Properties props = new Properties();
  props.put("bootstrap.servers", "master:9092,slave1:9092,slave2:9092");
  props.put("group.id", GROUPID);
  props.put("enable.auto.commit", "true");
  props.put("auto.commit.interval.ms", "1000");
  props.put("session.timeout.ms", "30000");
  props.put("max.poll.records", 1000);
  props.put("auto.offset.reset", "earliest");
  props.put("key.deserializer", StringDeserializer.class.getName());
  props.put("value.deserializer", StringDeserializer.class.getName());
  KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

由于我這是設(shè)置的自動(dòng)提交，所以消費(fèi)代碼如下:

我們需要先訂閱一個(gè)topic，也就是指定消費(fèi)哪一個(gè)topic。

consumer.subscribe(Arrays.asList(topic));

訂閱之后，我們?cè)購膋afka中拉取數(shù)據(jù):

ConsumerRecords<String, String> msgList=consumer.poll(1000);

一般來說進(jìn)行消費(fèi)會(huì)使用監(jiān)聽，這里我們就用for(;;)來進(jìn)行監(jiān)聽，并且設(shè)置消費(fèi)1000條就退出！

結(jié)果如下:

可以看到我們這里已經(jīng)成功消費(fèi)了生產(chǎn)的數(shù)據(jù)了。

代碼

那么生產(chǎn)者和消費(fèi)者的代碼如下:

生產(chǎn)者:

import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

/**
 * 
* Title: KafkaProducerTest
* Description: 
* kafka 生產(chǎn)者demo
* Version:1.0.0 
* @author pancm
* @date 2018年1月26日
 */
public class KafkaProducerTest implements Runnable {

 private final KafkaProducer<String, String> producer;
 private final String topic;
 public KafkaProducerTest(String topicName) {
  Properties props = new Properties();
  props.put("bootstrap.servers", "master:9092,slave1:9092,slave2:9092");
  props.put("acks", "all");
  props.put("retries", 0);
  props.put("batch.size", 16384);
  props.put("key.serializer", StringSerializer.class.getName());
  props.put("value.serializer", StringSerializer.class.getName());
  this.producer = new KafkaProducer<String, String>(props);
  this.topic = topicName;
 }

 @Override
 public void run() {
  int messageNo = 1;
  try {
   for(;;) {
    String messageStr="你好，這是第"+messageNo+"條數(shù)據(jù)";
    producer.send(new ProducerRecord<String, String>(topic, "Message", messageStr));
    //生產(chǎn)了100條就打印
    if(messageNo%100==0){
     System.out.println("發(fā)送的信息:" + messageStr);
    }
    //生產(chǎn)1000條就退出
    if(messageNo%1000==0){
     System.out.println("成功發(fā)送了"+messageNo+"條");
     break;
    }
    messageNo++;
   }
  } catch (Exception e) {
   e.printStackTrace();
  } finally {
   producer.close();
  }
 }
 
 public static void main(String args[]) {
  KafkaProducerTest test = new KafkaProducerTest("KAFKA_TEST");
  Thread thread = new Thread(test);
  thread.start();
 }
}

消費(fèi)者:

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


/**
 * 
* Title: KafkaConsumerTest
* Description: 
* kafka消費(fèi)者 demo
* Version:1.0.0 
* @author pancm
* @date 2018年1月26日
 */
public class KafkaConsumerTest implements Runnable {

 private final KafkaConsumer<String, String> consumer;
 private ConsumerRecords<String, String> msgList;
 private final String topic;
 private static final String GROUPID = "groupA";

 public KafkaConsumerTest(String topicName) {
  Properties props = new Properties();
  props.put("bootstrap.servers", "master:9092,slave1:9092,slave2:9092");
  props.put("group.id", GROUPID);
  props.put("enable.auto.commit", "true");
  props.put("auto.commit.interval.ms", "1000");
  props.put("session.timeout.ms", "30000");
  props.put("auto.offset.reset", "earliest");
  props.put("key.deserializer", StringDeserializer.class.getName());
  props.put("value.deserializer", StringDeserializer.class.getName());
  this.consumer = new KafkaConsumer<String, String>(props);
  this.topic = topicName;
  this.consumer.subscribe(Arrays.asList(topic));
 }

 @Override
 public void run() {
  int messageNo = 1;
  System.out.println("---------開始消費(fèi)---------");
  try {
   for (;;) {
     msgList = consumer.poll(1000);
     if(null!=msgList&&msgList.count()>0){
     for (ConsumerRecord<String, String> record : msgList) {
      //消費(fèi)100條就打印 ,但打印的數(shù)據(jù)不一定是這個(gè)規(guī)律的
      if(messageNo%100==0){
       System.out.println(messageNo+"=======receive: key = " + record.key() + ", value = " + record.value()+" offset==="+record.offset());
      }
      //當(dāng)消費(fèi)了1000條就退出
      if(messageNo%1000==0){
       break;
      }
      messageNo++;
     }
    }else{ 
     Thread.sleep(1000);
    }
   }  
  } catch (InterruptedException e) {
   e.printStackTrace();
  } finally {
   consumer.close();
  }
 } 
 public static void main(String args[]) {
  KafkaConsumerTest test1 = new KafkaConsumerTest("KAFKA_TEST");
  Thread thread1 = new Thread(test1);
  thread1.start();
 }
}

注: master、slave1、slave2 是因?yàn)槲以谧约旱沫h(huán)境做了關(guān)系映射，這個(gè)可以換成服務(wù)器的IP。

當(dāng)然項(xiàng)目我放在Github上了，有興趣的可以看看。 https://github.com/xuwujing/kafka （本地下載）

總結(jié)

簡單的開發(fā)一個(gè)kafka的程序需要以下步驟: