自动提交，手动提交（异步提交，同步提交，异步同步结合提交），原理：提交后，重新消费消息位移发生变化。
1 public class MyConsumer {
2
3 private static KafkaConsumer<String, String> consumer;
4 private static Properties properties;
5
6 static {
7
8 properties = new Properties();
9
10 properties.put("bootstrap.servers", "127.0.0.1:9092");
11 properties.put("key.deserializer",
12 "org.apache.kafka.common.serialization.StringDeserializer");
13 properties.put("value.deserializer",
14 "org.apache.kafka.common.serialization.StringDeserializer");
15 properties.put("group.id", "KafkaStudy");
16 }
17
18 private static void generalConsumeMessageAutoCommit() {
19
20 properties.put("enable.auto.commit", true);
21 consumer = new KafkaConsumer<>(properties);
22 consumer.subscribe(Collections.singleton("imooc-kafka-study-x"));
23
24 try {
25 while (true) {
26
27 boolean flag = true;
28 ConsumerRecords<String, String> records = consumer.poll(100);
29
30 for (ConsumerRecord<String, String> record : records) {
31 System.out.println(String.format(
32 "topic = %s, partition = %s, key = %s, value = %s",
33 record.topic(), record.partition(),
34 record.key(), record.value()
35 ));
36 if (record.value().equals("done")) {
37 flag = false;
38 }
39 }
40
41 if (!flag) {
42 break;
43 }
44 }
45 } finally {
46 consumer.close();
47 }
48 }
49
50 private static void generalConsumeMessageSyncCommit() {
51
52 properties.put("auto.commit.offset", false);
53 consumer = new KafkaConsumer<>(properties);
54 consumer.subscribe(Collections.singletonList("imooc-kafka-study-x"));
55
56 while (true) {
57 boolean flag = true;
58
59 ConsumerRecords<String, String> records =
60 consumer.poll(100);
61 for (ConsumerRecord<String, String> record : records) {
62 System.out.println(String.format(
63 "topic = %s, partition = %s, key = %s, value = %s",
64 record.topic(), record.partition(),
65 record.key(), record.value()
66 ));
67 if (record.value().equals("done")) {
68 flag = false;
69 }
70 }
71
72 try {
73 consumer.commitSync();
74 } catch (CommitFailedException ex) {
75 System.out.println("commit failed error: "
76 + ex.getMessage());
77 }
78
79 if (!flag) {
80 break;
81 }
82 }
83 }
84
85 private static void generalConsumeMessageAsyncCommit() {
86
87 properties.put("auto.commit.offset", false);
88 consumer = new KafkaConsumer<>(properties);
89 consumer.subscribe(Collections.singletonList("imooc-kafka-study-x"));
90
91 while (true) {
92 boolean flag = true;
93
94 ConsumerRecords<String, String> records =
95 consumer.poll(100);
96 for (ConsumerRecord<String, String> record : records) {
97 System.out.println(String.format(
98 "topic = %s, partition = %s, key = %s, value = %s",
99 record.topic(), record.partition(),
100 record.key(), record.value()
101 ));
102 if (record.value().equals("done")) {
103 flag = false;
104 }
105 }
106
107 // commit A, offset 2000
108 // commit B, offset 3000
109 consumer.commitAsync();
110
111 if (!flag) {
112 break;
113 }
114 }
115 }
116
117 private static void generalConsumeMessageAsyncCommitWithCallback() {
118
119 properties.put("auto.commit.offset", false);
120 consumer = new KafkaConsumer<>(properties);
121 consumer.subscribe(Collections.singletonList("imooc-kafka-study-x"));
122
123 while (true) {
124 boolean flag = true;
125
126 ConsumerRecords<String, String> records =
127 consumer.poll(100);
128 for (ConsumerRecord<String, String> record : records) {
129 System.out.println(String.format(
130 "topic = %s, partition = %s, key = %s, value = %s",
131 record.topic(), record.partition(),
132 record.key(), record.value()
133 ));
134 if (record.value().equals("done")) {
135 flag = false;
136 }
137 }
138
139 consumer.commitAsync((map, e) -> {
140 if (e != null) {
141 System.out.println("commit failed for offsets: " +
142 e.getMessage());
143 }
144 });
145
146 if (!flag) {
147 break;
148 }
149 }
150 }
151
152 @SuppressWarnings("all")
153 private static void mixSyncAndAsyncCommit() {
154
155 properties.put("auto.commit.offset", false);
156 consumer = new KafkaConsumer<>(properties);
157 consumer.subscribe(Collections.singletonList("imooc-kafka-study-x"));
158
159 try {
160
161 while (true) {
162 ConsumerRecords<String, String> records =
163 consumer.poll(100);
164
165 for (ConsumerRecord<String, String> record : records) {
166 System.out.println(String.format(
167 "topic = %s, partition = %s, key = %s, " +
168 "value = %s",
169 record.topic(), record.partition(),
170 record.key(), record.value()
171 ));
172 }
173
174 consumer.commitAsync();
175 }
176 } catch (Exception ex) {
177 System.out.println("commit async error: " + ex.getMessage());
178 } finally {
179 try {
180 consumer.commitSync();
181 } finally {
182 consumer.close();
183 }
184 }
185 }
186
187 public static void main(String[] args) {
188 generalConsumeMessageAutoCommit();
189 }
190 }
标签：演示,key,value,kafka,record,flag,提交,consumer,properties From： https://www.cnblogs.com/15078480385zyc/p/18170097