KafkaConsumer开发-创新互联
跟生产者一样,消费者也属于kafka的客户端,不过kafka消费者是从kafka读取数据的应用,侧重于读数据。一个或多个消费者订阅kafka集群中的topic,并从broker接收topic消息,从而进行业务处理。今天来学习下kafka consumer基本使用。
创新互联建站-专业网站定制、快速模板网站建设、高性价比镇海网站开发、企业建站全套包干低至880元,成熟完善的模板库,直接使用。一站式镇海网站制作公司更省心,省钱,快速模板网站建设找我们,业务覆盖镇海地区。费用合理售后完善,十载实体公司更值得信赖。消费者example 组件版本- kafka_2.13-3.3.1
- JDK17
- apache-maven-3.6.0
org.apache.kafka kafka-clients 3.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读取出来,并没有真正的处理),消息将丢失。它不会被再次读取,因为这些消息的偏移量已经提交。
- 批量拉取数据
- 消费者自动提交偏移量
- 对消息进行业务处理,如发送email,此时系统奔溃
- 系统重启后,从上次已提交的偏移量进行读取,在业务上造成消息丢失
在至少一次语义定义中,broker消息的每一个消息都会被传递到消费者,但是可能会存在重复拉取的场景,从而导致消息被重复处理。跟最多一次提交位置偏移量的时机不同,至少一次在处理消息后提交偏移量。
因此需要确保消息处理的幂等性,如对数据进行插入、更新操作;防止重复消费导致数据出现错乱。
至少一次消息处理的流程大致如下:
批量拉取数据
此时消费者并不提交偏移量
对消息进行业务处理
3.1 处理完成 提交偏移量 进行下一次拉取数据
3.2 消息处理失败(此时可能有一部分数据处理完成,还有一部分数据尚未处理)
重启应用 拉取数据,又会拉取之前的数据 导致消息被重复处理
有些场景不仅需要至少一次语义(保证数据不丢失),还需要精确一次语义。每条消息只投递一次,这需要消费者应用程序跟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://pwwzsj.com/article/pshec.html