Kafka 使用Java實現資料的生產和消費demo
前言
在上一篇中講述如何搭建kafka叢集,本篇則講述如何簡單的使用 kafka 。不過在使用kafka的時候,還是應該簡單的瞭解下kafka。
Kafka的介紹
Kafka是一種高吞吐量的分散式釋出訂閱訊息系統,它可以處理消費者規模的網站中的所有動作流資料。
Kafka 有如下特性:
- 以時間複雜度為O(1)的方式提供訊息持久化能力,即使對TB級以上資料也能保證常數時間複雜度的訪問效能。
- 高吞吐率。即使在非常廉價的商用機器上也能做到單機支援每秒100K條以上訊息的傳輸。
- 支援Kafka Server間的訊息分割槽,及分散式消費,同時保證每個Partition內的訊息順序傳輸。
- 同時支援離線資料處理和實時資料處理。
- Scale out:支援線上水平擴充套件。
kafka的術語
- Broker:Kafka叢集包含一個或多個伺服器,這種伺服器被稱為broker。
- Topic:每條釋出到Kafka叢集的訊息都有一個類別,這個類別被稱為Topic。(物理上不同Topic的訊息分開儲存,邏輯上一個Topic的訊息雖然保存於一個或多個broker上但使用者只需指定訊息的Topic即可生產或消費資料而不必關心資料存於何處)
- Partition:Partition是物理上的概念,每個Topic包含一個或多個Partition。
- Producer:負責釋出訊息到Kafka broker。
- Consumer:訊息消費者,向Kafka broker讀取訊息的客戶端。
- Consumer Group:每個Consumer屬於一個特定的Consumer Group(可為每個Consumer指定group name,若不指定group name則屬於預設的group)。
kafka核心Api
kafka有四個核心API
- 應用程式使用producer API釋出訊息到1個或多個topic中。
- 應用程式使用consumer API來訂閱一個或多個topic,並處理產生的訊息。
- 應用程式使用streams API充當一個流處理器,從1個或多個topic消費輸入流,併產生一個輸出流到1個或多個topic,有效地將輸入流轉換到輸出流。
- connector API允許構建或執行可重複使用的生產者或消費者,將topic連結到現有的應用程式或資料系統。
示例圖如下:
kafka 應用場景
- 構建可在系統或應用程式之間可靠獲取資料的實時流資料管道。
- 構建實時流應用程式,可以轉換或響應資料流。
以上介紹參考kafka官方文件。
開發準備
如果我們要開發一個kafka的程式,應該做些什麼呢?
首先,在搭建好kafka環境之後,我們要考慮的是我們是生產者還是消費者,也就是訊息的傳送者還是接受者。
不過在本篇中,生產者和消費者都會進行開發和講解。
在大致的瞭解kafka之後,我們來開發第一個程式。
這裡用的開發語言是Java,構建工具Maven。
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
在開發生產的時候,先簡單的介紹下kafka各種配置說明:
- bootstrap.servers: kafka的地址。
- acks:訊息的確認機制,預設值是0。
acks=0:如果設定為0,生產者不會等待kafka的響應。
acks=1:這個配置意味著kafka會把這條訊息寫到本地日誌檔案中,但是不會等待叢集中其他機器的成功響應。
acks=all:這個配置意味著leader會等待所有的follower同步完成。這個確保訊息不會丟失,除非kafka叢集中所有機器掛掉。這是最強的可用性保證。 - retries:配置為大於0的值的話,客戶端會在訊息傳送失敗時重新發送。
- batch.size:當多條訊息需要傳送到同一個分割槽時,生產者會嘗試合併網路請求。這會提高client和生產者的效率。
- key.serializer: 鍵序列化,預設org.apache.kafka.common.serialization.StringDeserializer。
- value.deserializer:值序列化,預設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的配置新增之後,我們便開始生產資料,生產資料程式碼只需如下就行:
producer.send(new ProducerRecord<String, String>(topic,key,value));
- topic: 訊息佇列的名稱,可以先行在kafka服務中進行建立。如果kafka中並未建立該topic,那麼便會自動建立!
- key:鍵值,也就是value對應的值,和Map類似。
value:要傳送的資料,資料格式為String型別的。
在寫好生產者程式之後,那我們先來生產吧!
我這裡傳送的訊息為:
String messageStr="你好,這是第"+messageNo+"條資料";
並且只發送1000條就退出,結果如下:
可以看到資訊成功的列印了。
如果不想用程式進行驗證程式是否傳送成功,以及訊息傳送的準確性,可以在kafka伺服器上使用命令檢視。
Kafka Consumer
kafka消費這塊應該來說是重點,畢竟大部分的時候,我們主要使用的是將資料進行消費。
kafka消費的配置如下:
- bootstrap.servers: kafka的地址。
- group.id:組名 不同組名可以重複消費。例如你先使用了組名A消費了kafka的1000條資料,但是你還想再次進行消費這1000條資料,並且不想重新去產生,那麼這裡你只需要更改組名就可以重複消費了。
- enable.auto.commit:是否自動提交,預設為true。
- auto.commit.interval.ms: 從poll(拉)的回話處理時長。
- session.timeout.ms:超時時間。
- max.poll.records:一次最大拉取的條數。
- auto.offset.reset:消費規則,預設earliest 。
earliest: 當各分割槽下有已提交的offset時,從提交的offset開始消費;無提交的offset時,從頭開始消費 。
latest: 當各分割槽下有已提交的offset時,從提交的offset開始消費;無提交的offset時,消費新產生的該分割槽下的資料 。
none: topic各分割槽都存在已提交的offset時,從offset後開始消費;只要有一個分割槽不存在已提交的offset,則丟擲異常。 - key.serializer: 鍵序列化,預設org.apache.kafka.common.serialization.StringDeserializer。
value.deserializer:值序列化,預設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);
由於我這是設定的自動提交,所以消費程式碼如下:
我們需要先訂閱一個topic,也就是指定消費哪一個topic。
consumer.subscribe(Arrays.asList(topic));
訂閱之後,我們再從kafka中拉取資料:
ConsumerRecords<String, String> msgList=consumer.poll(1000);
一般來說進行消費會使用監聽,這裡我們就用for(;;)來進行監聽, 並且設定消費1000條就退出!
結果如下:
可以看到我們這裡已經成功消費了生產的資料了。
程式碼
那麼生產者和消費者的程式碼如下:
生產者:
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 生產者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+"條資料";
producer.send(new ProducerRecord<String, String>(topic, "Message", messageStr));
//生產了100條就列印
if(messageNo%100==0){
System.out.println("傳送的資訊:" + messageStr);
}
//生產1000條就退出
if(messageNo%1000==0){
System.out.println("成功傳送了"+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();
}
}
消費者:
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消費者 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("---------開始消費---------");
try {
for (;;) {
msgList = consumer.poll(1000);
if(null!=msgList&&msgList.count()>0){
for (ConsumerRecord<String, String> record : msgList) {
//消費100條就列印 ,但列印的資料不一定是這個規律的
if(messageNo%100==0){
System.out.println(messageNo+"=======receive: key = " + record.key() + ", value = " + record.value()+" offset==="+record.offset());
}
//當消費了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 是因為我在自己的環境做了關係對映,這個可以換成伺服器的IP。
總結
簡單的開發一個kafka的程式需要以下步驟:
- 成功搭建kafka伺服器,併成功啟動!
- 得到kafka服務資訊,然後在程式碼中進行相應的配置。
- 配置完成之後,監聽kafka中的訊息佇列是否有訊息產生。
- 將產生的資料進行業務邏輯處理!
到此,本文就結束了,謝謝閱讀!