理解 Kafka poll()、flush() 和 commit()答案

【问题标题】：Understanding Kafka poll(), flush() & commit()理解 Kafka poll()、flush() 和 commit()
【发布时间】：2019-11-23 10:25:09
【问题描述】：

我是 Kafka 新手，正在为我的新应用程序尝试一些小用例。用例基本上是， Kafka-producer—>Kafka-Consumer—>flume-Kafka source—>flume-hdfs-sink.

消费时（步骤2），以下是步骤顺序.. 1.消费者投票（1.0） 1.a.生产多个主题（多个水槽代理正在监听） 1.b。生产。轮询（） 2. Flush() 每 25 个消息 3. Commit() 每个消息 (asynchCommit=false)

问题1：这个动作顺序对吗！？！

问题2：这是否会导致任何数据丢失，因为刷新是每 25 个消息，并且提交是针对每个消息？！？

问题3：poll() for producer和poll()consumer的区别？

问题4：提交消息但未刷新时会发生什么！？！

如果有人能帮助我理解生产者/消费者之间用于轮询、刷新和提交的偏移示例，我将不胜感激。

提前致谢！！

【问题讨论】：

标签： apache-kafka kafka-consumer-api kafka-producer-api flume

【解决方案1】：

让我们先简单了解一下卡夫卡：

什么是kafka生产者：

t.turner@devs:~/developers/softwares/kafka_2.12-2.2.0$ bin/kafka-console-producer.sh --broker-list 100.102.1.40:9092,100.102.1.41:9092 --topic company_wallet_db_v3-V3_0_0-transactions
>{"created_at":1563415200000,"payload":{"action":"insert","entity":{"amount":40.0,"channel":"INTERNAL","cost_rate":1.0,"created_at":"2019-07-18T02:00:00Z","currency_id":1,"direction":"debit","effective_rate":1.0,"explanation":"Voucher,"exchange_rate":null,expired","id":1563415200,"instrument":null,"instrument_id":null,"latitude":null,"longitude":null,"other_party":null,"primary_account_id":2,"receiver_phone":null,"secondary_account_id":362,"sequence":1,"settlement_id":null,"status":"success","type":"voucher_expiration","updated_at":"2019-07-18T02:00:00Z","primary_account_previous_balance":0.0,"secondary_account_previous_balance":0.0}},"track_id":"a011ad33-2cdd-48a5-9597-5c27c8193033"}
[2019-07-21 11:53:37,907] WARN [Producer clientId=console-producer] Error while fetching metadata with correlation id 7 : {company_wallet_db_v3-V3_0_0-transactions=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)

您可以忽略警告。似乎 Kafka 找不到主题并自动创建主题。

让我们看看kafka是如何存储这个消息的：

生产者在代理服务器中创建一个目录/kafka-logs（对于 apache kafka）或/kafka-cf-data（对于 confluent 版本）

drwxr-xr-x   2 root root  4096 Jul 21 08:53 company_wallet_db_v3-V3_0_0-transactions-0

cd 进入这个目录，然后列出文件。您将看到存储实际数据的.log 文件：

-rw-r--r--   1 root root 10485756 Jul 21 08:53 00000000000000000000.timeindex
-rw-r--r--   1 root root 10485760 Jul 21 08:53 00000000000000000000.index
-rw-r--r--   1 root root        8 Jul 21 08:53 leader-epoch-checkpoint
drwxr-xr-x   2 root root     4096 Jul 21 08:53 .
-rw-r--r--   1 root root      762 Jul 21 08:53 00000000000000000000.log

如果你打开日志文件，你会看到：

^@^@^@^@^@^@^@^@^@^@^Bî^@^@^@^@^B<96>T<88>ò^@^@^@^@^@^@^@^@^Al^S<85><98>k^@^@^Al^S<85><98>kÿÿÿÿÿÿÿÿÿÿÿÿÿÿ^@^@^@^Aö
^@^@^@^Aè
{"created_at":1563415200000,"payload":{"action":"insert","entity":{"amount":40.0,"channel":"INTERNAL","cost_rate":1.0,"created_at":"2019-07-18T02:00:00Z","currency_id":1,"direction":"debit","effective_rate":1.0,"explanation":"Voucher,"exchange_rate":null,expired","id":1563415200,"instrument":null,"instrument_id":null,"latitude":null,"longitude":null,"other_party":null,"primary_account_id":2,"receiver_phone":null,"secondary_account_id":362,"sequence":1,"settlement_id":null,"status":"success","type":"voucher_expiration","updated_at":"2019-07-18T02:00:00Z","primary_account_previous_balance":0.0,"secondary_account_previous_balance":0.0}},"track_id":"a011ad33-2cdd-48a5-9597-5c27c8193033"}^@

让我们了解消费者如何轮询和读取记录：

什么是卡夫卡民意调查：

Kafka 为分区中的每条记录维护一个数字偏移量。此偏移量充当该记录中的唯一标识符分区，也表示消费者在划分。例如，位置 5 的消费者消费了偏移量为 0 到 4 的记录，接下来将收到带有偏移量 5. 实际上有两个位置概念与消费者的用户：消费者的位置给出的偏移量将发出的下一条记录。它会比消费者在该分区中看到的最高偏移量。它每次消费者收到消息时自动前进调用 poll(long)。

因此，poll 将持续时间作为输入，读取该持续时间的 00000000000000000000.log 文件，并将它们返回给消费者。

什么时候删除消息：

Kafka 负责消息的刷新。有两种方式：

基于时间：默认为 7 天。可以改变使用 log.retention.ms=1680000
基于大小：可以设置为 log.retention.bytes=10487500

现在让我们看看消费者：

t.turner@devs:~/developers/softwares/kafka_2.12-2.2.0$ bin/kafka-console-consumer.sh --bootstrap-server 100.102.1.40:9092 --topic company_wallet_db_v3-V3_0_0-transactions --from-beginning
{"created_at":1563415200000,"payload":{"action":"insert","entity":{"amount":40.0,"channel":"INTERNAL","cost_rate":1.0,"created_at":"2019-07-18T02:00:00Z","currency_id":1,"direction":"debit","effective_rate":1.0,"explanation":"Voucher,"exchange_rate":null,expired","id":1563415200,"instrument":null,"instrument_id":null,"latitude":null,"longitude":null,"other_party":null,"primary_account_id":2,"receiver_phone":null,"secondary_account_id":362,"sequence":1,"settlement_id":null,"status":"success","type":"voucher_expiration","updated_at":"2019-07-18T02:00:00Z","primary_account_previous_balance":0.0,"secondary_account_previous_balance":0.0}},"track_id":"a011ad33-2cdd-48a5-9597-5c27c8193033"}
^CProcessed a total of 1 messages

上述命令指示消费者从offset = 0 读取。 Kafka 为这个控制台消费者分配一个group_id 并维护这个group_id 读取的最后一个偏移量。所以，它可以将更新的消息推送到这个consumer-group

什么是 Kafka Commit：

Commit 是一种告诉 kafka 消费者已成功处理的消息的方式。这可以被认为是更新group-id : current_offset + 1 之间的查找。您可以使用使用者对象的 commitAsync() 或 commitSync() 方法来管理它。

参考：https://kafka.apache.org/10/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html

【讨论】：