前言
在上一篇中講述如何搭建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的依賴如下:
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
<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配置如下:
|
1
2
3
4
5
6
7
8
|
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的配置添加之后，我們便開始生產數據，生產數據代碼只需如下就行:
|
1
|
producer.send(new producerrecord<string, string>(topic,key,value)); |
- topic: 消息隊列的名稱，可以先行在kafka服務中進行創建。如果kafka中并未創建該topic，那么便會自動創建！
- key:鍵值，也就是value對應的值，和map類似。
- value:要發送的數據，數據格式為string類型的。
在寫好生產者程序之后，那我們先來生產吧！
我這里發送的消息為:
|
1
|
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配置如下:
|
1
2
3
4
5
6
7
8
9
10
11
|
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。
|
1
|
consumer.subscribe(arrays.aslist(topic)); |
訂閱之后，我們再從kafka中拉取數據:
|
1
|
consumerrecords<string, string> msglist=consumer.poll(1000); |
一般來說進行消費會使用監聽，這里我們就用for(;;)來進行監聽， 并且設置消費1000條就退出！
結果如下:
可以看到我們這里已經成功消費了生產的數據了。
代碼
那么生產者和消費者的代碼如下:
生產者:
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
|
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(); }} |
消費者:
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
|
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
總結
以上就是這篇文章的全部內容了，希望本文的內容對大家的學習或者工作具有一定的參考學習價值，如果有疑問大家可以留言交流，謝謝大家對服務器之家的支持。
原文鏈接：https://www.cnblogs.com/xuwujing/p/8371127.html
