blog/BigData-Notes
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

kafka消费者

Browse Source
This commit is contained in:
luoxiang
2019-05-29 00:01:32 +08:00
parent 80f5886d01
commit 4be48146c9
14 changed files with 407 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
code/Kafka/kafka-basis/src/main/java/com/heibaiying/consumers/ConsumerASyn.java
View File

@@ -20,6 +20,7 @@ public class ConsumerASyn {
Properties props = new Properties();
props.put("bootstrap.servers", "hadoop001:9092");
props.put("group.id", group);
props.put("enable.auto.commit", false);
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);

1
code/Kafka/kafka-basis/src/main/java/com/heibaiying/consumers/ConsumerASynAndSyn.java
View File

@@ -20,6 +20,7 @@ public class ConsumerASynAndSyn {
Properties props = new Properties();
props.put("bootstrap.servers", "hadoop001:9092");
props.put("group.id", group);
props.put("enable.auto.commit", false);
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);

1
code/Kafka/kafka-basis/src/main/java/com/heibaiying/consumers/ConsumerASynWithOffsets.java
View File

@@ -24,6 +24,7 @@ public class ConsumerASynWithOffsets {
Properties props = new Properties();
props.put("bootstrap.servers", "hadoop001:9092");
props.put("group.id", group);
props.put("enable.auto.commit", false);
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);

1
code/Kafka/kafka-basis/src/main/java/com/heibaiying/consumers/ConsumerExit.java
View File

@@ -23,6 +23,7 @@ public class ConsumerExit {
Properties props = new Properties();
props.put("bootstrap.servers", "hadoop001:9092");
props.put("group.id", group);
props.put("enable.auto.commit", false);
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);

1
code/Kafka/kafka-basis/src/main/java/com/heibaiying/consumers/ConsumerGroup.java
View File

@@ -22,6 +22,7 @@ public class ConsumerGroup {
props.put("bootstrap.servers", "hadoop001:9092");
/*指定分组ID*/
props.put("group.id", group);
props.put("enable.auto.commit", true);
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);

1
code/Kafka/kafka-basis/src/main/java/com/heibaiying/consumers/ConsumerSyn.java
View File

@@ -20,6 +20,7 @@ public class ConsumerSyn {
Properties props = new Properties();
props.put("bootstrap.servers", "hadoop001:9092");
props.put("group.id", group);
props.put("enable.auto.commit", false);
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);

3
code/Kafka/kafka-basis/src/main/java/com/heibaiying/consumers/RebalanceListener.java
View File

@@ -15,6 +15,7 @@ public class RebalanceListener {
Properties props = new Properties();
props.put("bootstrap.servers", "hadoop001:9092");
props.put("group.id", group);
props.put("enable.auto.commit", false);
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);
@@ -27,7 +28,7 @@ public class RebalanceListener {
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
System.out.println("再均衡即将触发");
// 提交当前偏移量
// 提交已经处理的偏移量
consumer.commitSync(offsets);
}

1
code/Kafka/kafka-basis/src/main/java/com/heibaiying/consumers/StandaloneConsumer.java
View File

@@ -25,6 +25,7 @@ public class StandaloneConsumer {
Properties props = new Properties();
props.put("bootstrap.servers", "hadoop001:9092");
props.put("group.id", group);
props.put("enable.auto.commit", false);
props.put("key.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(props);