KafkaConsumer开发-创新互联

Kafka Consumer - 消费者

跟生产者一样，消费者也属于kafka的客户端，不过kafka消费者是从kafka读取数据的应用，侧重于读数据。一个或多个消费者订阅kafka集群中的topic，并从broker接收topic消息，从而进行业务处理。今天来学习下kafka consumer基本使用。

创新互联建站-专业网站定制、快速模板网站建设、高性价比镇海网站开发、企业建站全套包干低至880元,成熟完善的模板库,直接使用。一站式镇海网站制作公司更省心,省钱,快速模板网站建设找我们，业务覆盖镇海地区。费用合理售后完善，十载实体公司更值得信赖。消费者example 组件版本

• kafka_2.13-3.3.1
• JDK17
• apache-maven-3.6.0

Maven依赖

org.apache.kafkakafka-clients3.3.1

消费者代码

public static void main(String[] args){String topicName = "consumer-topic";

        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "app_01");
        props.put("enable.auto.commit", true);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,   StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        KafkaConsumerconsumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList(topicName));
        try {while (true) {ConsumerRecordsrecords = consumer.poll(Duration.ofSeconds(1));
                records.forEach(record ->{System.out.println("Message received " + record.value());
                });
            }
        }finally {consumer.close();
        }
    }

测试验证

• 创建topic

  ./bin/kafka-topics.sh --create --topic consumer-topic --bootstrap-server localhost:9092

• 启动生产者 - 这里使用kafka自带的生产者脚本进行测试

  ./bin/kafka-console-producer.sh --topic consumer-topic --bootstrap-server localhost:9092

• 测试结果

  

至此 一个简单的kafka消费者程序已经开发完成，代码不多，开发起来也快。但是关于kafka 消费者内部有很多的原理、细节需要去梳理，否则出现问题就会茫然失措，不知所以。

pull VS poll

上面的消费者程序有一个很核心的细节需要关注，即kafka 消费者以什么的方式对数据进行消费。对比其他传统的消息中间件，消息消费的方式主要有两种：

• 推送模式 - broker 主动推送消息给消费者
• 拉取模式- 消费者主动从broker拉取消息

kafka在设计之处，就考虑这个问题：消费者从broker拉取数据，还是broker主动推送数据给消费者。在这方面kafka采用更为传统的设计：消费者主动拉取，其优势如下：

• 拉取模式 可以根据消费者自身的消费能力对数据处理。如生产者大量数据，消费者消费能力有限
• 拉取模式 消费者可以根据实际情况对数据进行批量处理。推送模式很难做到这一点
• broker被设计成无状态模式，broker不需要对记录每一个消费者的偏移量，由客户端自己控制 便于kafka集群扩展

消息传递语义

在介绍消息传递语义之前，首先要了解下kafka 消费者位置（也叫做偏移量）管理。

位移管理

kafka 消费者端需要为每个读取的topic 分区保存消费进度，即当前分区中消费者消费消息的最新位置。该位置也叫做偏移量- offset。消费者需要定期地想kafka提交自己的位置信息，实际上，偏移量通常是下一条待消费消息的位置。如下图

从kafka broker读取消息，开发者可以选择提交偏移量的时间，消费者默认自动提交偏移量，这可能会带来一些风险。

最多一次

在这种情况下，在调用poll（）后，一旦收到消息批，就立即提交偏移量。如果后续处理失败（如业务处理过程中发生异常，数据只是被从Broker读取出来，并没有真正的处理），消息将丢失。它不会被再次读取，因为这些消息的偏移量已经提交。

1. 批量拉取数据
2. 消费者自动提交偏移量
3. 对消息进行业务处理，如发送email，此时系统奔溃
4. 系统重启后，从上次已提交的偏移量进行读取，在业务上造成消息丢失

至少一次

在至少一次语义定义中，broker消息的每一个消息都会被传递到消费者，但是可能会存在重复拉取的场景，从而导致消息被重复处理。跟最多一次提交位置偏移量的时机不同，至少一次在处理消息后提交偏移量。

因此需要确保消息处理的幂等性，如对数据进行插入、更新操作；防止重复消费导致数据出现错乱。

至少一次消息处理的流程大致如下：

1. 批量拉取数据

2. 此时消费者并不提交偏移量

3. 对消息进行业务处理

   3.1 处理完成 提交偏移量 进行下一次拉取数据

   3.2 消息处理失败（此时可能有一部分数据处理完成，还有一部分数据尚未处理）

4. 重启应用 拉取数据，又会拉取之前的数据 导致消息被重复处理

精确一次

有些场景不仅需要至少一次语义(保证数据不丢失)，还需要精确一次语义。每条消息只投递一次，这需要消费者应用程序跟kafka相互配合、相互合作就可以实现精确一次语义

• 使用kafka事务API实现精确一次语义
• 对于消费者应用程序，要有效地实现一次，必须使用幂等性消费

位移配置

props.put("enable.auto.commit", true);

enable.auto.commit 参数默认值为true，kafka默认在后台线程中周期性的提交消费者偏移量

auto.commit.interval.ms默认为5秒，如果enable.auto.commit参数设置为true，即消费者5秒提交一次位移。

在至少一次、精确一次语义中 需要将该参数设置为false，由应用程序手动提交偏移量

//...
props.put("enable.auto.commit", false);
//...
while (true) {ConsumerRecordsrecords = consumer.poll(Duration.ofSeconds(1));
  records.forEach(record ->{System.out.println("Message received " + record.value());
  });
  //提交偏移量
  consumer.commitSync();
}

根据不用的应用场景，kakfa提供了多个API让开发者对消费者位移进行手动管理

auto.offset.reset

指定消费者从broker拉取数据的位置，有以下几个选项可以配置

• earliest - 从最开始进行消费
• latest - 从最后消费的偏移量进行消费 默认值
• none - 如果未找到使用者组的先前偏移量，则向使用者抛出异常

消费者组

props.put(ConsumerConfig.GROUP_ID_CONFIG, "app_01");

在开发kafka消费者代码时，必须指定消费者组，否则会报错，那么该参数有什么作用呢。在回答这个问题之前，先假设两个应用场景

• kafka中消息特别多，需要增加消费者加快消息处理的速度，避免出现消息堆积
• 某一类消息特别重要，需要被多个应用程序同时消费 - 如购买商品的消息，需要被库存应用、积分应用同时消费

借用RocketMQ中的概念（个人觉得比较合适），以上两种应用场景叫做集群消费、广播消费

• 集群消费 - 多个消费者共同消费某一个主题内的消息
• 广播消费 - 每一个消息被多个消费者同时消费

kafka 内部以消费者组的方式实现以上两点要求

• 同一个消费组的不同消费实例 共同消费topiic的消息
• 同一个消息被不同的消费组同时消费

在开发代码时，只需要按需更改一下配置即可

props.put(ConsumerConfig.GROUP_ID_CONFIG, "app_B");
props.put("client.id", "client_02");

你是否还在寻找稳定的海外服务器提供商？创新互联www.cdcxhl.cn海外机房具备T级流量清洗系统配攻击溯源，准确流量调度确保服务器高可用性，企业级服务器适合批量采购，新人活动首月15元起，快前往官网查看详情吧

本文名称：KafkaConsumer开发-创新互联
浏览地址：http://bjjierui.cn/article/pshec.html