本文介绍kafka-SDK的消费者消费消息示例的使用

准备配置

安装Java依赖库

在pom.xml中添加以下依赖。

  1. <dependency>
  2.     <groupId>org.apache.kafka</groupId>
  3.     <artifactId>kafka-clients</artifactId>
  4.     <version>2.8.0</version>
  5. </dependency>

使用示例

推荐使用安服优的AfuKafkaConfig默认配置，需要先引入相关sdk详情可参照：SDK下载与使用，可以直接使用以下示例进行单Consumer订阅消息。
以下是创建单Consumer订阅消息程序AfuKafkaConsumerDemo.java。

  1. import com.afu.common.sdk.config.AfuKafkaConfig;
  3. import org.apache.kafka.clients.consumer.ConsumerConfig;
  4. import org.apache.kafka.clients.consumer.ConsumerRecord;
  5. import org.apache.kafka.clients.consumer.ConsumerRecords;
  6. import org.apache.kafka.clients.consumer.KafkaConsumer;
  7. import org.apache.kafka.clients.producer.ProducerConfig;
  10. import java.time.Duration;
  11. import java.util.*;
  14. public class AfuKafkaConsumerDemo {
  17.     public static void main(String args[]) {
  19.         AfuKafkaConfig kafkaConfig=new AfuKafkaConfig();
  20.         Properties props = new Properties();
  22.         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConfig.getBOOTSTRAP_SERVERS_CONFIG());
  23.         //两次Poll之间的最大允许间隔。
  24.         //消费者超过该值没有返回心跳，服务端判断消费者处于非存活状态，服务端将消费者从Consumer Group移除并触发Rebalance。
  25.         //(AfuKafkaConfig配置为20(s))
  26.         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, kafkaConfig.getSESSION_TIMEOUT_MS_CONFIG());
  27.         //每次Poll的最大数量。
  28.         //注意该值不要改得太大，如果Poll太多数据，而不能在下次Poll之前消费完，则会触发一次负载均衡，产生卡顿。
  29.         //(AfuKafkaConfig配置为20)
  30.         props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, kafkaConfig.getMAX_POLL_RECORDS_CONFIG());
  32.         //消息的反序列化方式。
  33.         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, kafkaConfig.getKEY_DESERIALIZER_CLASS_CONFIG());
  34.         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, kafkaConfig.getVALUE_DESERIALIZER_CLASS_CONFIG());
  37.         //当前消费实例所属的消费组，请在控制台申请之后填写。
  38.         //属于同一个组的消费实例，会负载消费消息。
  39.         //(AfuKafkaConfig配置为"default_consumer_group")
  40.         props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaConfig.getGROUP_ID_CONFIG());
  42.                 //构造消费对象，也即生成一个消费实例。
  43.         KafkaConsumer<String, String> consumer = new org.apache.kafka.clients.consumer.KafkaConsumer<>(props);
  45.         //设置消费组订阅的消费组id（groupId），可以订阅多个。
  46.         //如果GROUP_ID_CONFIG是一样，则订阅的groupId也建议设置成一样。
  47.         List<String> subscribedGroupIds =  new ArrayList<>();
  49.         //如果需要订阅多个GroupId，则在这里添加进去即可。
  50.         //每个GroupId需要先在安服优物联云平台进行创建,您可以查看后再填入。
  51.         subscribedGroupIds.add("TestGroupId");
  52.         consumer.subscribe(subscribedGroupIds);
  54.         //循环消费消息。
  55.         while (true){
  56.             try {
  57.                 ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
  58.                 //必须在下次Poll之前消费完这些数据, 且总耗时不得超过SESSION_TIMEOUT_MS_CONFIG。
  59.                 //建议开一个单独的线程池来消费消息，然后异步返回结果。
  60.                 for (ConsumerRecord<String, String> record : records) {
  61.                     System.out.println(String.format("Consume partition:%d offset:%d", record.partition(), record.offset())+"value:"+record.value());
  62.                 }
  63.             } catch (Exception e) {
  64.                 try {
  65.                     Thread.sleep(1000);
  66.                 } catch (Throwable ignore) {
  68.                 }
  69.                 e.printStackTrace();
  70.             }
  71.         }
  72.     }
  73. }

当生产者发布消息后，该demo将进行单Consumer的消费