java kafka 消费_java编程之Kafka_消费者API详解
1
消息發(fā)送
1、異步發(fā)送導(dǎo)入依賴
org.apache.kafka
kafka-clients
0.11.0.0
編寫代碼
需要用到的類：
KafkaProducer：需要創(chuàng)建一個生產(chǎn)者對象，用來發(fā)送數(shù)據(jù)
ProducerConfig：獲取所需的一系列配置參數(shù)
ProducerRecord：每條數(shù)據(jù)都要封裝成一個ProducerRecord對象
不帶回調(diào)函數(shù)的API
package com.heima.kafka;
import org.apache.kafka.clients.producer.*;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
public class CustomProducer {
public static void main(String[] args) throws ExecutionException, InterruptedException {
Properties props = new Properties();
props.put("bootstrap.servers", "hadoop102:9092");//kafka集群，broker-list
props.put("acks", "all");
props.put("retries", 1);//重試次數(shù)
props.put("batch.size", 16384);//批次大小
props.put("linger.ms", 1);//等待時間
props.put("buffer.memory", 33554432);//RecordAccumulator緩沖區(qū)大小
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
Producer producer = new KafkaProducer<>(props);
for (int i = 0; i < 100; i++) {
producer.send(new ProducerRecord("first", Integer.toString(i), Integer.toString(i)));
}
producer.close();
}
}
帶回調(diào)函數(shù)的API
回調(diào)函數(shù)會在producer收到ack時調(diào)用，為異步調(diào)用，該方法有兩個參數(shù)，分別是RecordMetadata和Exception，如果Exception為null，說明消息發(fā)送成功，如果Exception不為null，說明消息發(fā)送失敗。
注意：消息發(fā)送失敗會自動重試，不需要我們在回調(diào)函數(shù)中手動重試。
package com.heima.kafka;
import org.apache.kafka.clients.producer.*;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
public class CustomProducer {
public static void main(String[] args) throws ExecutionException, InterruptedException {
Properties props = new Properties();
props.put("bootstrap.servers", "hadoop102:9092");//kafka集群，broker-list
props.put("acks", "all");
props.put("retries", 1);//重試次數(shù)
props.put("batch.size", 16384);//批次大小
props.put("linger.ms", 1);//等待時間
props.put("buffer.memory", 33554432);//RecordAccumulator緩沖區(qū)大小
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
Producer producer = new KafkaProducer<>(props);
for (int i = 0; i < 100; i++) {
producer.send(new ProducerRecord("first", Integer.toString(i), Integer.toString(i)), new Callback() {
//回調(diào)函數(shù)，該方法會在Producer收到ack時調(diào)用，為異步調(diào)用
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
if (exception == null) {
System.out.println("success->" + metadata.offset());
} else {
exception.printStackTrace();
}
}
});
}
producer.close();
}
}
2
同步發(fā)送
同步發(fā)送的意思就是，一條消息發(fā)送之后，會阻塞當(dāng)前線程，直至返回ack。
由于send方法返回的是一個Future對象，根據(jù)Futrue對象的特點(diǎn)，我們也可以實(shí)現(xiàn)同步發(fā)送的效果，只需在調(diào)用Future對象的get方即可。
package com.heima.kafka;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
public class CustomProducer {
public static void main(String[] args) throws ExecutionException, InterruptedException {
Properties props = new Properties();
props.put("bootstrap.servers", "hadoop102:9092");//kafka集群，broker-list
props.put("acks", "all");
props.put("retries", 1);//重試次數(shù)
props.put("batch.size", 16384);//批次大小
props.put("linger.ms", 1);//等待時間
props.put("buffer.memory", 33554432);//RecordAccumulator緩沖區(qū)大小
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
Producer producer = new KafkaProducer<>(props);
for (int i = 0; i < 100; i++) {
producer.send(new ProducerRecord("first", Integer.toString(i), Integer.toString(i))).get();
}
producer.close();
}
}
3
消息消費(fèi)
Consumer消費(fèi)數(shù)據(jù)時的可靠性是很容易保證的，因?yàn)閿?shù)據(jù)在Kafka中是持久化的，故不用擔(dān)心數(shù)據(jù)丟失問題。
由于consumer在消費(fèi)過程中可能會出現(xiàn)斷電宕機(jī)等故障，consumer恢復(fù)后，需要從故障前的位置的繼續(xù)消費(fèi)，所以consumer需要實(shí)時記錄自己消費(fèi)到了哪個offset，以便故障恢復(fù)后繼續(xù)消費(fèi)。所以offset的維護(hù)是Consumer消費(fèi)數(shù)據(jù)是必須考慮的問題。
1、自動提交offset導(dǎo)入依賴
org.apache.kafka
kafka-clients
0.11.0.0
編寫代碼
需要用到的類：
KafkaConsumer：需要創(chuàng)建一個消費(fèi)者對象，用來消費(fèi)數(shù)據(jù)
ConsumerConfig：獲取所需的一系列配置參數(shù)
ConsuemrRecord：每條數(shù)據(jù)都要封裝成一個ConsumerRecord對象
為了使我們能夠?qū)Ｗ⒂谧约旱臉I(yè)務(wù)邏輯，Kafka提供了自動提交offset的功能。
自動提交offset的相關(guān)參數(shù)：
enable.auto.commit：是否開啟自動提交offset功能
auto.commit.interval.ms：自動提交offset的時間間隔
以下為自動提交offset的代碼：
package com.heima.kafka;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Arrays;
import java.util.Properties;
public class CustomConsumer {
public static void main(String[] args) {
Properties props = new Properties();
props.put("bootstrap.servers", "hadoop102:9092");
props.put("group.id", "test");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("first"));
while (true) {
ConsumerRecords records = consumer.poll(100);
for (ConsumerRecord record : records)
System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}
}
}
2、手動提交offset
雖然自動提交offset十分簡潔便利，但由于其是基于時間提交的，開發(fā)人員難以把握offset提交的時機(jī)。因此Kafka還提供了手動提交offset的API。
手動提交offset的方法有兩種：分別是commitSync(同步提交)和commitAsync(異步提交)。兩者的相同點(diǎn)是，都會將本次poll的一批數(shù)據(jù)最高的偏移量提交；不同點(diǎn)是，commitSync阻塞當(dāng)前線程，一直到提交成功，并且會自動失敗重試(由不可控因素導(dǎo)致，也會出現(xiàn)提交失敗)；而commitAsync則沒有失敗重試機(jī)制，故有可能提交失敗。
同步提交offset
由于同步提交offset有失敗重試機(jī)制，故更加可靠，以下為同步提交offset的示例。
package com.heima.kafka.consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Arrays;
import java.util.Properties;
/**
* @author liubo
*/
public class CustomComsumer {
public static void main(String[] args) {
Properties props = new Properties();
props.put("bootstrap.servers", "hadoop102:9092");//Kafka集群
props.put("group.id", "test");//消費(fèi)者組，只要group.id相同，就屬于同一個消費(fèi)者組
props.put("enable.auto.commit", "false");//關(guān)閉自動提交offset
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("first"));//消費(fèi)者訂閱主題
while (true) {
ConsumerRecords records = consumer.poll(100);//消費(fèi)者拉取數(shù)據(jù)
for (ConsumerRecord record : records) {
System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}
consumer.commitSync();//同步提交，當(dāng)前線程會阻塞知道offset提交成功
}
}
}
異步提交offset
雖然同步提交offset更可靠一些，但是由于其會阻塞當(dāng)前線程，直到提交成功。因此吞吐量會收到很大的影響。因此更多的情況下，會選用異步提交offset的方式。
以下為異步提交offset的示例：
package com.heima.kafka.consumer;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import java.util.Arrays;
import java.util.Map;
import java.util.Properties;
/**
* @author liubo
*/
public class CustomConsumer {
public static void main(String[] args) {
Properties props = new Properties();
props.put("bootstrap.servers", "hadoop102:9092");//Kafka集群
props.put("group.id", "test");//消費(fèi)者組，只要group.id相同，就屬于同一個消費(fèi)者組
props.put("enable.auto.commit", "false");//關(guān)閉自動提交offset
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("first"));//消費(fèi)者訂閱主題
while (true) {
ConsumerRecords records = consumer.poll(100);//消費(fèi)者拉取數(shù)據(jù)
for (ConsumerRecord record : records) {
System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}
consumer.commitAsync(new OffsetCommitCallback() {
@Override
public void onComplete(Map offsets, Exception exception) {
if (exception != null) {
System.err.println("Commit failed for" + offsets);
}
}
});//異步提交
}
}
}
3、自定義存儲offset
Kafka 0.9版本之前，offset存儲在zookeeper，0.9版本之后，默認(rèn)將offset存儲在Kafka的一個內(nèi)置的topic中。除此之外，Kafka還可以選擇自定義存儲offset。
Offset的維護(hù)是相當(dāng)繁瑣的，因?yàn)樾枰紤]到消費(fèi)者的Rebalace。
當(dāng)有新的消費(fèi)者加入消費(fèi)者組、已有的消費(fèi)者推出消費(fèi)者組或者所訂閱的主題的分區(qū)發(fā)生變化，就會觸發(fā)到分區(qū)的重新分配，重新分配的過程叫做Rebalance。
消費(fèi)者發(fā)生Rebalance之后，每個消費(fèi)者消費(fèi)的分區(qū)就會發(fā)生變化。因此消費(fèi)者要首先獲取到自己被重新分配到的分區(qū)，并且定位到每個分區(qū)最近提交的offset位置繼續(xù)消費(fèi)。
要實(shí)現(xiàn)自定義存儲offset，需要借助ConsumerRebalanceListener，以下為示例代碼，其中提交和獲取offset的方法，需要根據(jù)所選的offset存儲系統(tǒng)自行實(shí)現(xiàn)。
package com.heima.kafka.consumer;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import java.util.*;
/**
* @author liubo
*/
public class CustomConsumer {
private static Map currentOffset = new HashMap<>();
public static void main(String[] args) {
Properties props = new Properties();
props.put("bootstrap.servers", "hadoop102:9092");//Kafka集群
props.put("group.id", "test");//消費(fèi)者組，只要group.id相同，就屬于同一個消費(fèi)者組
props.put("enable.auto.commit", "false");//關(guān)閉自動提交offset
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("first"), new ConsumerRebalanceListener() {
//該方法會在Rebalance之前調(diào)用
@Override
public void onPartitionsRevoked(Collection partitions) {
commitOffset(currentOffset);
}
//該方法會在Rebalance之后調(diào)用
@Override
public void onPartitionsAssigned(Collection partitions) {
currentOffset.clear();
for (TopicPartition partition : partitions) {
consumer.seek(partition, getOffset(partition));//定位到最近提交的offset位置繼續(xù)消費(fèi)
}
}
});
while (true) {
ConsumerRecords records = consumer.poll(100);//消費(fèi)者拉取數(shù)據(jù)
for (ConsumerRecord record : records) {
System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
currentOffset.put(new TopicPartition(record.topic(), record.partition()), record.offset());
}
commitOffset(currentOffset);
}
}
//獲取某分區(qū)的最新offset
private static long getOffset(TopicPartition partition) {
return 0;
}
//提交該消費(fèi)者所有分區(qū)的offset
private static void commitOffset(Map currentOffset) {
}
}
總結(jié)
以上是生活随笔為你收集整理的java kafka 消费_java编程之Kafka_消费者API详解的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： php与mysql店事物处理_PHP与M
- 下一篇： 诸神战纪2java_诸神战纪2死神归来