理解Kafka民意测验（），flush（）

Question

我刚接触Kafka，正在为我的新应用程序尝试一些小用例。用例基本上是Kafka制作人-

当消费时（步骤2），下面是步骤的顺序...1.消费者。轮询（1.0）1. a.产生多个主题（多个水槽代理正在监听）1.b。产生。轮询（）2。每25个msgs刷新（）3。提交（）每个msgs（asynchCommit=false）

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

问题2：这会导致数据丢失吗？因为刷新是每25毫秒一次，提交是每一毫秒一次？！？

问题3：生产者和消费者之间的区别？

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

如果有人能帮助我理解生产者/消费者之间的偏移示例，我将非常感激。

提前谢谢！！

Answer 1

让我们先简单地了解一下Kafka：

什么是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（对于合流版本）创建一个目录

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民调：

Kafka为分区中的每条记录维护一个数字偏移量。该偏移量充当该分区内记录的唯一标识符，还表示使用者在分区中的位置。例如，位于位置5的消费者已经消费了偏移量为0到4的记录，接下来将收到偏移量为5的记录。实际上，与消费者的用户相关的位置有两个概念：消费者的位置给出下一条记录的偏移量。它将比消费者在该分区中看到的最高偏移量大一倍。每次消费者在call to poll（长）中收到消息时，它都会自动前进。

因此，轮询将一个持续时间作为输入，读取该持续时间的00000000000000000000.log文件，并将其返回给消费者。

何时删除邮件：

Kafka负责信息的刷新。有两种方法：

1. 基于时间：默认为7天。可以使用log.retention.ms=1680000
更改
2. 基于大小：可以设置为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

上面的命令指示使用者从偏移=0读取。Kafka为这个控制台使用者分配一个group_id，并维护这个group_id读取的最后一个偏移量。因此，它可以将更新的消息推送到这个消费者组

什么是Kafkapromise：

提交是告知Kafka消费者已成功处理的消息的一种方式。这可以被认为是在组id:current_offset 1之间更新查找。您可以使用consumer对象的commitSync（）或commitSync（）方法来管理它。

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