前言
在上一篇中講述如何搭建kafka集群，本篇則講述如何簡單的使用 kafka 。不過在使用kafka的時候，還是應(yīng)該簡單的了解下kafka。
kafka的介紹
kafka是一種高吞吐量的分布式發(fā)布訂閱消息系統(tǒng)，它可以處理消費者規(guī)模的網(wǎng)站中的所有動作流數(shù)據(jù)。
kafka 有如下特性：
- 以時間復(fù)雜度為o(1)的方式提供消息持久化能力，即使對tb級以上數(shù)據(jù)也能保證常數(shù)時間復(fù)雜度的訪問性能。
- 高吞吐率。即使在非常廉價的商用機器上也能做到單機支持每秒100k條以上消息的傳輸。
- 支持kafka server間的消息分區(qū)，及分布式消費，同時保證每個partition內(nèi)的消息順序傳輸。
- 同時支持離線數(shù)據(jù)處理和實時數(shù)據(jù)處理。
- scale out：支持在線水平擴展。
kafka的術(shù)語
- broker：kafka集群包含一個或多個服務(wù)器，這種服務(wù)器被稱為broker。
- topic：每條發(fā)布到kafka集群的消息都有一個類別，這個類別被稱為topic。（物理上不同topic的消息分開存儲，邏輯上一個topic的消息雖然保存于一個或多個broker上但用戶只需指定消息的topic即可生產(chǎn)或消費數(shù)據(jù)而不必關(guān)心數(shù)據(jù)存于何處）
- partition：partition是物理上的概念，每個topic包含一個或多個partition。
- producer：負(fù)責(zé)發(fā)布消息到kafka broker。
- consumer：消息消費者，向kafka broker讀取消息的客戶端。
- consumer group:每個consumer屬于一個特定的consumer group（可為每個consumer指定group name，若不指定group name則屬于默認(rèn)的group）。
kafka核心api
kafka有四個核心api
- 應(yīng)用程序使用producer api發(fā)布消息到1個或多個topic中。
- 應(yīng)用程序使用consumer api來訂閱一個或多個topic,并處理產(chǎn)生的消息。
- 應(yīng)用程序使用streams api充當(dāng)一個流處理器,從1個或多個topic消費輸入流,并產(chǎn)生一個輸出流到1個或多個topic,有效地將輸入流轉(zhuǎn)換到輸出流。
- connector api允許構(gòu)建或運行可重復(fù)使用的生產(chǎn)者或消費者,將topic鏈接到現(xiàn)有的應(yīng)用程序或數(shù)據(jù)系統(tǒng)。
示例圖如下:
kafka 應(yīng)用場景
- 構(gòu)建可在系統(tǒng)或應(yīng)用程序之間可靠獲取數(shù)據(jù)的實時流數(shù)據(jù)管道。
- 構(gòu)建實時流應(yīng)用程序，可以轉(zhuǎn)換或響應(yīng)數(shù)據(jù)流。
以上介紹參考kafka官方文檔。
開發(fā)準(zhǔn)備
如果我們要開發(fā)一個kafka的程序，應(yīng)該做些什么呢？
首先，在搭建好kafka環(huán)境之后，我們要考慮的是我們是生產(chǎn)者還是消費者，也就是消息的發(fā)送者還是接受者。
不過在本篇中，生產(chǎn)者和消費者都會進(jìn)行開發(fā)和講解。
在大致的了解kafka之后，我們來開發(fā)第一個程序。
這里用的開發(fā)語言是java，構(gòu)建工具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
在開發(fā)生產(chǎn)的時候，先簡單的介紹下kafka各種配置說明：
- bootstrap.servers： kafka的地址。
- acks:消息的確認(rèn)機制，默認(rèn)值是0。
- acks=0：如果設(shè)置為0，生產(chǎn)者不會等待kafka的響應(yīng)。
- acks=1：這個配置意味著kafka會把這條消息寫到本地日志文件中，但是不會等待集群中其他機器的成功響應(yīng)。
- acks=all：這個配置意味著leader會等待所有的follower同步完成。這個確保消息不會丟失，除非kafka集群中所有機器掛掉。這是最強的可用性保證。
- retries：配置為大于0的值的話，客戶端會在消息發(fā)送失敗時重新發(fā)送。
- batch.size:當(dāng)多條消息需要發(fā)送到同一個分區(qū)時，生產(chǎn)者會嘗試合并網(wǎng)絡(luò)請求。這會提高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配置如下:
|
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的配置添加之后，我們便開始生產(chǎn)數(shù)據(jù)，生產(chǎn)數(shù)據(jù)代碼只需如下就行:
|
1
|
producer.send(new producerrecord<string, string>(topic,key,value)); |
- topic: 消息隊列的名稱，可以先行在kafka服務(wù)中進(jìn)行創(chuàng)建。如果kafka中并未創(chuàng)建該topic，那么便會自動創(chuàng)建！
- key:鍵值，也就是value對應(yīng)的值，和map類似。
- value:要發(fā)送的數(shù)據(jù)，數(shù)據(jù)格式為string類型的。
在寫好生產(chǎn)者程序之后，那我們先來生產(chǎn)吧！
我這里發(fā)送的消息為:
|
1
|
string messagestr="你好，這是第"+messageno+"條數(shù)據(jù)"; |
并且只發(fā)送1000條就退出，結(jié)果如下:
可以看到信息成功的打印了。
如果不想用程序進(jìn)行驗證程序是否發(fā)送成功，以及消息發(fā)送的準(zhǔn)確性，可以在kafka服務(wù)器上使用命令查看。
kafka consumer
kafka消費這塊應(yīng)該來說是重點，畢竟大部分的時候，我們主要使用的是將數(shù)據(jù)進(jìn)行消費。
kafka消費的配置如下:
- bootstrap.servers： kafka的地址。
- group.id：組名 不同組名可以重復(fù)消費。例如你先使用了組名a消費了kafka的1000條數(shù)據(jù)，但是你還想再次進(jìn)行消費這1000條數(shù)據(jù)，并且不想重新去產(chǎn)生，那么這里你只需要更改組名就可以重復(fù)消費了。
- enable.auto.commit：是否自動提交，默認(rèn)為true。
- auto.commit.interval.ms: 從poll(拉)的回話處理時長。
- session.timeout.ms:超時時間。
- max.poll.records:一次最大拉取的條數(shù)。
-
auto.offset.reset：消費規(guī)則，默認(rèn)earliest 。
earliest: 當(dāng)各分區(qū)下有已提交的offset時，從提交的offset開始消費；無提交的offset時，從頭開始消費 。
latest: 當(dāng)各分區(qū)下有已提交的offset時，從提交的offset開始消費；無提交的offset時，消費新產(chǎn)生的該分區(qū)下的數(shù)據(jù) 。
none: topic各分區(qū)都存在已提交的offset時，從offset后開始消費；只要有一個分區(qū)不存在已提交的offset，則拋出異常。 - key.serializer: 鍵序列化，默認(rèn)org.apache.kafka.common.serialization.stringdeserializer。
- value.deserializer:值序列化，默認(rèn)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); |
由于我這是設(shè)置的自動提交，所以消費代碼如下:
我們需要先訂閱一個topic，也就是指定消費哪一個topic。
|
1
|
consumer.subscribe(arrays.aslist(topic)); |
訂閱之后，我們再從kafka中拉取數(shù)據(jù):
|
1
|
consumerrecords<string, string> msglist=consumer.poll(1000); |
一般來說進(jìn)行消費會使用監(jiān)聽，這里我們就用for(;;)來進(jìn)行監(jiān)聽， 并且設(shè)置消費1000條就退出！
結(jié)果如下:
可以看到我們這里已經(jīng)成功消費了生產(chǎn)的數(shù)據(jù)了。
代碼
那么生產(chǎn)者和消費者的代碼如下:
生產(chǎn)者:
|
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 生產(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(); }} |
消費者:
|
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條就打印 ,但打印的數(shù)據(jù)不一定是這個規(guī)律的 if(messageno%100==0){ system.out.println(messageno+"=======receive: key = " + record.key() + ", value = " + record.value()+" offset==="+record.offset()); } //當(dāng)消費了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 是因為我在自己的環(huán)境做了關(guān)系映射，這個可以換成服務(wù)器的ip。
當(dāng)然項目我放在github上了，有興趣的可以看看。https://github.com/xuwujing/kafka
總結(jié)
簡單的開發(fā)一個kafka的程序需要以下步驟:
- 成功搭建kafka服務(wù)器，并成功啟動！
- 得到kafka服務(wù)信息，然后在代碼中進(jìn)行相應(yīng)的配置。
- 配置完成之后，監(jiān)聽kafka中的消息隊列是否有消息產(chǎn)生。
- 將產(chǎn)生的數(shù)據(jù)進(jìn)行業(yè)務(wù)邏輯處理！
kafka介紹參考官方文檔:http://kafka.apache.org/intro
總結(jié)
以上就是這篇文章的全部內(nèi)容了，希望本文的內(nèi)容對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，如果有疑問大家可以留言交流，謝謝大家對服務(wù)器之家的支持。
原文鏈接：https://www.cnblogs.com/xuwujing/p/8371127.html
