之前使用kafka的KafkaStream，讓每個(gè)消費(fèi)者和對(duì)應(yīng)的patition建立對(duì)應(yīng)的流來(lái)讀取kafka上面的數(shù)據(jù)，如果comsumer得到數(shù)據(jù)，那么kafka就會(huì)自動(dòng)去維護(hù)該comsumer的offset，例如在獲取到kafka的消息后正準(zhǔn)備入庫(kù)（未入庫(kù)），但是消費(fèi)者掛了，那么如果讓kafka自動(dòng)去維護(hù)offset，它就會(huì)認(rèn)為這條數(shù)據(jù)已經(jīng)被消費(fèi)了，那么會(huì)造成數(shù)據(jù)丟失。
但是kafka可以讓你自己去手動(dòng)提交，如果在上面的場(chǎng)景中，那么需要我們手動(dòng)commit，如果comsumer掛了 那么程序就不會(huì)執(zhí)行commit這樣的話 其他同group的消費(fèi)者又可以消費(fèi)這條數(shù)據(jù)，保證數(shù)據(jù)不丟，先要做如下設(shè)置：
|
1
2
|
//設(shè)置不自動(dòng)提交，自己手動(dòng)更新offsetproperties.put("enable.auto.commit", "false"); |
使用如下api提交：
|
1
|
consumer.commitSync(); |
注意：
剛做了個(gè)測(cè)試，如果我從kafka中取出5條數(shù)據(jù)，分別為1,2,3,4,5，如果消費(fèi)者在執(zhí)行一些邏輯在執(zhí)行1,2,3，4的時(shí)候都失敗了未提交commit，然后消費(fèi)5做邏輯成功了提交了commit，那么offset也會(huì)被移動(dòng)到5那一條數(shù)據(jù)那里，1,2,3,4 相當(dāng)于也會(huì)丟失
如果是做消費(fèi)者取出數(shù)據(jù)執(zhí)行一些操作，全部都失敗的話，然后重啟消費(fèi)者，這些數(shù)據(jù)會(huì)從失敗的時(shí)候重新開始讀取
所以消費(fèi)者還是應(yīng)該自己做容錯(cuò)機(jī)制
測(cè)試項(xiàng)目結(jié)構(gòu)如下:
其中ConsumerThreadNew類：
|
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
|
package com.lijie.kafka;import java.util.ArrayList;import java.util.Arrays;import java.util.List;import org.apache.kafka.clients.consumer.ConsumerRecord;import org.apache.kafka.clients.consumer.ConsumerRecords;import org.apache.kafka.clients.consumer.KafkaConsumer;import org.slf4j.Logger;import org.slf4j.LoggerFactory;/** * * * @Filename ConsumerThreadNew.java * * @Description * * @Version 1.0 * * @Author Lijie * * @Email [email protected] * * @History *<li>Author: Lijie</li> *<li>Date: 2017年3月21日</li> *<li>Version: 1.0</li> *<li>Content: create</li> * */public class ConsumerThreadNew implements Runnable { private static Logger LOG = LoggerFactory.getLogger(ConsumerThreadNew.class); //KafkaConsumer kafka生產(chǎn)者 private KafkaConsumer<String, String> consumer; //消費(fèi)者名字 private String name; //消費(fèi)的topic組 private List<String> topics; //構(gòu)造函數(shù) public ConsumerThreadNew(KafkaConsumer<String, String> consumer, String topic, String name) { super(); this.consumer = consumer; this.name = name; this.topics = Arrays.asList(topic); } @Override public void run() { consumer.subscribe(topics); List<ConsumerRecord<String, String>> buffer = new ArrayList<>(); // 批量提交數(shù)量 final int minBatchSize = 1; while (true) { ConsumerRecords<String, String> records = consumer.poll(100); for (ConsumerRecord<String, String> record : records) { LOG.info("消費(fèi)者的名字為:" + name + ",消費(fèi)的消息為：" + record.value()); buffer.add(record); } if (buffer.size() >= minBatchSize) { //這里就是處理成功了然后自己手動(dòng)提交 consumer.commitSync(); LOG.info("提交完畢"); buffer.clear(); } } }} |
MyConsume類如下：
|
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
|
package com.lijie.kafka;import java.util.Properties;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import org.apache.kafka.clients.consumer.KafkaConsumer;import org.slf4j.Logger;import org.slf4j.LoggerFactory;/** * * * @Filename MyConsume.java * * @Description * * @Version 1.0 * * @Author Lijie * * @Email [email protected] * * @History *<li>Author: Lijie</li> *<li>Date: 2017年3月21日</li> *<li>Version: 1.0</li> *<li>Content: create</li> * */public class MyConsume { private static Logger LOG = LoggerFactory.getLogger(MyConsume.class); public MyConsume() { // TODO Auto-generated constructor stub } public static void main(String[] args) { Properties properties = new Properties(); properties.put("bootstrap.servers", "10.0.4.141:19093,10.0.4.142:19093,10.0.4.143:19093"); //設(shè)置不自動(dòng)提交，自己手動(dòng)更新offset properties.put("enable.auto.commit", "false"); properties.put("auto.offset.reset", "latest"); properties.put("zookeeper.connect", "10.0.4.141:2181,10.0.4.142:2181,10.0.4.143:2181"); properties.put("session.timeout.ms", "30000"); properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); properties.put("group.id", "lijieGroup"); properties.put("zookeeper.connect", "192.168.80.123:2181"); properties.put("auto.commit.interval.ms", "1000"); ExecutorService executor = Executors.newFixedThreadPool(5); //執(zhí)行消費(fèi) for (int i = 0; i < 7; i++) { executor.execute(new ConsumerThreadNew(new KafkaConsumer<String, String>(properties), "lijietest", "消費(fèi)者" + (i + 1))); } }} |
MyProducer類如下：
|
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
|
package com.lijie.kafka;import java.util.Properties;import org.apache.kafka.clients.producer.KafkaProducer;import org.apache.kafka.clients.producer.ProducerRecord;/** * * * @Filename MyProducer.java * * @Description * * @Version 1.0 * * @Author Lijie * * @Email [email protected] * * @History *<li>Author: Lijie</li> *<li>Date: 2017年3月21日</li> *<li>Version: 1.0</li> *<li>Content: create</li> * */public class MyProducer { private static Properties properties; private static KafkaProducer<String, String> pro; static { //配置 properties = new Properties(); properties.put("bootstrap.servers", "10.0.4.141:19093,10.0.4.142:19093,10.0.4.143:19093"); //序列化類型 properties .put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); //創(chuàng)建生產(chǎn)者 pro = new KafkaProducer<>(properties); } public static void main(String[] args) throws Exception { produce("lijietest"); } public static void produce(String topic) throws Exception { //模擬message // String value = UUID.randomUUID().toString(); for (int i = 0; i < 10000; i++) { //封裝message ProducerRecord<String, String> pr = new ProducerRecord<String, String>(topic, i + ""); //發(fā)送消息 pro.send(pr); Thread.sleep(1000); } }} |
pom文件如下：
|
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
|
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>lijie-kafka-offset</groupId> <artifactId>lijie-kafka-offset</artifactId> <version>0.0.1-SNAPSHOT</version> <dependencies> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka_2.11</artifactId> <version>0.10.1.1</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-common</artifactId> <version>2.2.0</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-hdfs</artifactId> <version>2.2.0</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>2.2.0</version> </dependency> <dependency> <groupId>org.apache.hbase</groupId> <artifactId>hbase-client</artifactId> <version>1.0.3</version> </dependency> <dependency> <groupId>org.apache.hbase</groupId> <artifactId>hbase-server</artifactId> <version>1.0.3</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-hdfs</artifactId> <version>2.2.0</version> </dependency> <dependency> <groupId>jdk.tools</groupId> <artifactId>jdk.tools</artifactId> <version>1.7</version> <scope>system</scope> <systemPath>${JAVA_HOME}/lib/tools.jar</systemPath> </dependency> <dependency> <groupId>org.apache.httpcomponents</groupId> <artifactId>httpclient</artifactId> <version>4.3.6</version> </dependency> </dependencies> <build> <plugins> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-compiler-plugin</artifactId> <configuration> <source>1.7</source> <target>1.7</target> </configuration> </plugin> </plugins> </build></project> |
補(bǔ)充：kafka javaAPI 手動(dòng)維護(hù)偏移量
我就廢話不多說(shuō)了，大家還是直接看代碼吧~
|
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
|
package com.kafka;import kafka.javaapi.PartitionMetadata;import kafka.javaapi.consumer.SimpleConsumer;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.clients.consumer.OffsetAndMetadata;import org.apache.kafka.common.TopicPartition;import org.junit.Test;import java.util.*;public class ConsumerManageOffet {//broker的地址，//與老版的kafka的區(qū)別是，新版本的kafka把偏移量保存到了broker，而老版本的是把偏移量保存到了zookeeper中//所以在讀取數(shù)據(jù)時(shí)，應(yīng)當(dāng)設(shè)置broker的地址 private static String ips = "192.168.136.150:9092,192.168.136.151:9092,192.168.136.152:9092"; public static void main(String[] args) { Properties props = new Properties(); props.put("bootstrap.servers",ips); props.put("group.id","test02"); props.put("auto.offset.reset","earliest"); props.put("max.poll.records","10"); props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer"); KafkaConsumer<String,String> consumer = new KafkaConsumer<>(props); consumer.subscribe(Arrays.asList("my-topic")); System.out.println("---------------------"); while(true){ ConsumerRecords<String,String> records = consumer.poll(10); System.out.println("+++++++++++++++++++++++"); for(ConsumerRecord<String,String> record: records){ System.out.println("---"); System.out.printf("offset=%d,key=%s,value=%s%n",record.offset(), record.key(),record.value()); } } } //手動(dòng)維護(hù)偏移量 @Test public void autoManageOffset2(){ Properties props = new Properties(); //broker的地址 props.put("bootstrap.servers",ips); //這是消費(fèi)者組 props.put("group.id","groupPP"); //設(shè)置消費(fèi)的偏移量，如果以前消費(fèi)過(guò)則接著消費(fèi)，如果沒有就從頭開始消費(fèi) props.put("auto.offset.reset","earliest"); //設(shè)置自動(dòng)提交偏移量為false props.put("enable.auto.commit","false"); //設(shè)置Key和value的序列化 props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); //new一個(gè)消費(fèi)者 KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); //指定消費(fèi)的topic consumer.subscribe(Arrays.asList("my-topic")); while(true){ ConsumerRecords<String, String> records = consumer.poll(1000); //通過(guò)records獲取這個(gè)集合中的數(shù)據(jù)屬于那幾個(gè)partition Set<TopicPartition> partitions = records.partitions(); for(TopicPartition tp : partitions){ //通過(guò)具體的partition把該partition中的數(shù)據(jù)拿出來(lái)消費(fèi) List<ConsumerRecord<String, String>> partitionRecords = records.records(tp); for(ConsumerRecord r : partitionRecords){ System.out.println(r.offset() +" "+r.key()+" "+r.value()); } //獲取新這個(gè)partition中的最后一條記錄的offset并加1 那么這個(gè)位置就是下一次要提交的offset long newOffset = partitionRecords.get(partitionRecords.size() - 1).offset() + 1; consumer.commitSync(Collections.singletonMap(tp,new OffsetAndMetadata(newOffset))); } } }} |
以上為個(gè)人經(jīng)驗(yàn)，希望能給大家一個(gè)參考，也希望大家多多支持服務(wù)器之家。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教。
原文鏈接：https://blog.csdn.net/qq_20641565/article/details/64440425
