Kafka 使用Java實現數據的生產和消費demo
阿新 • • 發佈：2018-01-28
== 輸入流 是你 www esc 驗證 rri else 分開
前言
在上一篇中講述如何搭建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。
當然項目我放在Github上了，有興趣的可以看看。 https://github.com/xuwujing/kafka
總結
簡單的開發一個kafka的程序需要以下步驟:
- 成功搭建kafka服務器，並成功啟動！
- 得到kafka服務信息，然後在代碼中進行相應的配置。
- 配置完成之後，監聽kafka中的消息隊列是否有消息產生。
- 將產生的數據進行業務邏輯處理！
kafka介紹參考官方文檔:
http://kafka.apache.org/intro
到此，本文就結束了，謝謝閱讀！
Kafka 使用Java實現數據的生產和消費demo