Kafka Producer Consumer

1、kafka producer consumerproducer api ducer.kafkaproducer 假如想學(xué)習(xí)java工程化、高性能及分布式、深化淺出。微服務(wù)、spring，mybatis，netty源碼分析的伴侶可以加我的java高級溝通：854630135，群里有阿里大牛直播講解技術(shù)，以及java大型互聯(lián)網(wǎng)技術(shù)的視頻免費共享給大家。 1 props.put("bootstrap.servers", "28:9092"); 2 pr

2、ops.put("acks", "all"); 3 props.put("retries", 0); 4 props.put("batch.size", 16384); 5 props.put("linger.ms", 1); 6 props.put("buffer.memory", 33554432); 7 props.put("key.serializer&a

3、mp;quot;, "mon.serialization.stringserializer"); 8 props.put("value.serializer", "mon.serialization.stringserializer"); 9 10 producer producer = new kafkaproducer(props); 11 for (int i = 0; i ("foo", integer.tostring(i), integer

4、.tostring(i), new callback() 13 override 14 public void oncompletion(recordmetadata recordmetadata, exception e) 15 if (null != e) 16 e.printstacktrace(); 17 else 18 system.out.println("callback: " + recordmetadata.topic() + " " + recordmetadata.offset(); 19 20 21

5、 ); 22 23 producer.close(); producer由一個緩沖池組成，這個緩沖池中維護著那些還沒有被傳送到服務(wù)器上的記錄，而且有一個后臺的i/o線程負責(zé)將這些記錄轉(zhuǎn)換為哀求并將其傳送到集群上去。 send()辦法是異步的。當(dāng)調(diào)用它以后就把記錄放到buffer中并立刻返回。這就允許生產(chǎn)者批量的發(fā)送記錄。 acks配置項控制的是完成的標(biāo)準(zhǔn)，即什么樣的哀求被認為是完成了的。本例中其值設(shè)置的是"all"表示客戶端會等待直到全部記錄徹低被提交，這是最慢的一種方式也是持久化最好的一種方式。 假如哀求失敗了，生產(chǎn)者可以自動重試。由于這里我們設(shè)置retr

6、ies為0，所以它不重試。 生產(chǎn)者對每個分區(qū)都維護了一個buffers，其中放的是未被發(fā)送的記錄。這些buffers的大小是通過batch.size配置項來控制的。 默認狀況下，即使一個buffer還有未用法的空間（ps：buffer沒滿）也會立刻發(fā)送。假如你想要削減哀求的次數(shù)，你可以設(shè)置linger.ms為一個大于0的數(shù)。這個命令將告知生產(chǎn)者在發(fā)送哀求之前先等待多少毫秒，以希翼能有更多的記錄到達好填滿buffer。在本例中，我們設(shè)置的是1毫秒，表示我們的哀求將會延遲1毫秒發(fā)送，這樣做是為了等待更多的記錄到達，1毫秒之后即使buffer沒有被填滿，哀求也會發(fā)送。（ps：略微說明一下這段話，pr

7、oducer調(diào)用send()辦法只是將記錄放到buffer中，然后由一個后臺線程將buffer中的記錄傳送到服務(wù)器上。這里所說的哀求指的是從buffer到服務(wù)器。默認狀況下記錄被放到buffer以后立刻被發(fā)送到服務(wù)器，為了削減哀求服務(wù)器的次數(shù)，可以通過設(shè)置linger.ms，這個配置項表示等多少毫秒以后再發(fā)送，這樣做是希翼每次哀求可以發(fā)送更多的記錄，以此削減哀求次數(shù)） 假如想學(xué)習(xí)java工程化、高性能及分布式、深化淺出。微服務(wù)、spring，mybatis，netty源碼分析的伴侶可以加我的java高級溝通：854630135，群里有阿里大牛直播講解技術(shù)，以及java大型互聯(lián)網(wǎng)技術(shù)的視頻免費共

8、享給大家。 buffer.memory控制的是總的buffer內(nèi)存數(shù)量 key.serializer 和 value.serializer表示怎樣將key和value對象轉(zhuǎn)成字節(jié) 從kafka 0.11開頭，kafkaproducer支持兩種模型：the idempotent producer and the transactional producer（冪等producer和事務(wù)producer）。冪等producer強調(diào)的是起碼一次精確的投遞。事務(wù)producer允許應(yīng)用程序原子的發(fā)送消息到多個分區(qū)或者主題。 為了啟用冪等性，必需將enable.idempotence這個配置的值設(shè)為tru

9、e。假如你這樣設(shè)置了，那么retries默認是integer.max_value，并且acks默認是all。為了利用冪等producer的優(yōu)勢，請避開應(yīng)用程序級別的重新發(fā)送。 為了用法事務(wù)producer，你必需配置transactional.id。假如transactional.id被設(shè)置，冪等性自動被啟用。 1 properties props = new properties(); 2 props.put("bootstrap.servers", "28:9092"); 3 props.put(

10、"transactional.id", "my-transactional-id"); 4 5 producer producer = new kafkaproducer(props, new stringserializer(), new stringserializer(); 6 7 producer.inittransactions(); 8 9 try 10 producer.begintransaction(); 11 12 for (int i = 11; i ("bar",

11、 integer.tostring(i), integer.tostring(i); 14 15 / this method will flush any unsent records before actually committing the transaction 16 mittransaction(); 17 catch (producerfencedexception | outofordersequenceexception | authorizationexception e) 18 producer.close(); 19 catch (kafkaexception e) 20

12、 / by calling producer.aborttransaction() upon receiving a kafkaexception we can ensure 21 / that any successful writes are marked as aborted, hence keeping the transactional guarantees. 22 producer.aborttransaction(); 23 24 25 producer.close(); 假如想學(xué)習(xí)java工程化、高性能及分布式、深化淺出。微服務(wù)、spring，mybatis，netty源碼分析

13、的伴侶可以加我的java高級溝通：854630135，群里有阿里大牛直播講解技術(shù)，以及java大型互聯(lián)網(wǎng)技術(shù)的視頻免費共享給大家。 consumer api org.apache.kafka.clients.consumer.kafkaconsumer offsets and consumer position 對于分區(qū)中的每條記錄，kafka維護一個數(shù)值偏移量。這個偏移量是分區(qū)中一條記錄的唯一標(biāo)識，同時也是消費者在分區(qū)中的位置。例如，一個消費者在分區(qū)中的位置是5，表示它已經(jīng)消費了偏移量從0到4的記錄，并且接下來它將消費偏移量為5的記錄。相對于消費者用戶來說，這里事實上有兩個位置的概念。 消費

14、者的position表示下一條將要消費的記錄的offset。每次消費者通過調(diào)用poll(long)接收消息的時候這個position會自動增強。 committed position表示已經(jīng)被存儲的最后一個偏移量。消費者可以自動的周期性提交offsets，也可以通過調(diào)用提交api(e.g. commitsync and commitasync)手動的提交position。 consumer groups and topic subscriptions kafka用"consumer groups"（消費者組）的概念來允許一組進程分開處理和消費記錄。這些處理在

15、同一個機器上舉行，也可以在不同的機器上。同一個消費者組中的消費者實例有相同的group.id 組中的每個消費者可以動態(tài)設(shè)置它們想要訂閱的主題列表。kafka給每個訂閱的消費者組都投遞一份消息。這歸功于消費者組中全部成員之間的均衡分區(qū)，以至于每個分區(qū)都可以被指定到組中精確的一個消費者。假設(shè)一個主題有4個分區(qū)，一個組中有2個消費者，那么每個消費者將處理2個分區(qū)。 消費者組中的成員是動態(tài)維護的：假如一個消費者處理失敗了，那么分配給它的分區(qū)將會被重新分給組中其它消費者。 在概念上，你可以把一個消費者組想象成一個單個的規(guī)律訂閱者，并且每個規(guī)律訂閱者由多個進程組成。作為一個多訂閱系統(tǒng)，kafka天生就支持

16、對于給定的主題可以有隨意數(shù)量的消費者組。 automatic offset committing 1 properties props = new properties(); 2 props.put("bootstrap.servers", "28:9092"); 3 props.put("group.id", "test"); 4 props.put("mit", "true

17、"); 5 props.put("erval.ms", "1000"); 6 props.put("key.deserializer", "mon.serialization.stringdeserializer"); 7 props.put("value.deserializer", "mon.serialization.stringdeserializer&

18、;quot;); 8 kafkaconsumer consumer = new kafkaconsumer(props); 9 consumer.subscribe(arrays.aslist("foo", "bar"); 10 while (true) 11 consumerrecords records = consumer.poll(100); 12 for (consumerrecord record : records) 13 system.out.printf("offset = %d, ke

19、y = %s, value = %s%n", record.offset(), record.key(), record.value(); 14 15 設(shè)置mit意味著自動提交已消費的記錄的offset manual offset control 代替消費者周期性的提交已消費的offsets，用戶可以控制什么時候記錄被認為是已經(jīng)消費并提交它們的offsets。 1 properties props = new properties(); 2 props.put("bootstrap.servers", "localhost:9092"); 3 props.put("group.id", "test"); 4 props.put("mit", "false"); 5 props.put("key.