我正在使用apache kafka进行消息传递。我用Java实现了生产者和使用者。我们如何获取主题中的消息数量?
从消费者的角度来看,唯一想到这一点的方法是实际使用消息并计算它们。
Kafka代理公开JMX计数器,显示自启动以来收到的消息数量,但您无法知道已经清除了多少消息。
在大多数常见情况下,Kafka中的消息最好被视为无限流,并且获得当前保留在磁盘上的数量的离散值是不相关的。此外,当处理一个代理集群时,事情会变得更加复杂,这些代理在一个主题中都有一部分消息。
运行以下(假设kafka-console-consumer.sh
在路径上):
kafka-console-consumer.sh --from-beginning \
--bootstrap-server yourbroker:9092 --property print.key=true \
--property print.value=false --property print.partition \
--topic yourtopic --timeout-ms 5000 | tail -n 10|grep "Processed a total of"
使用Kafka 2.11-1.0.0的Java客户端,您可以执行以下操作:
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("test"));
while(true) {
ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records) {
System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
// after each message, query the number of messages of the topic
Set<TopicPartition> partitions = consumer.assignment();
Map<TopicPartition, Long> offsets = consumer.endOffsets(partitions);
for(TopicPartition partition : offsets.keySet()) {
System.out.printf("partition %s is at %d\n", partition.topic(), offsets.get(partition));
}
}
}
输出是这样的:
offset = 10, key = null, value = un
partition test is at 13
offset = 11, key = null, value = deux
partition test is at 13
offset = 12, key = null, value = trois
partition test is at 13
摘录自Kafka文档
弃权在0.9.0.0
kafka-consumer-offset-checker.sh(kafka.tools.ConsumerOffsetChecker)已被弃用。展望未来,请使用kafka-consumer-groups.sh(kafka.admin.ConsumerGroupCommand)来实现此功能。
我正在运行Kafka代理,为服务器和客户端启用了SSL。我用下面的命令
kafka-consumer-groups.sh --bootstrap-server Broker_IP:Port --list --command-config /tmp/ssl_config
kafka-consumer-groups.sh --bootstrap-server Broker_IP:Port --command-config /tmp/ssl_config --describe --group group_name_x
其中/ tmp / ssl_config如下所示
security.protocol=SSL
ssl.truststore.location=truststore_file_path.jks
ssl.truststore.password=truststore_password
ssl.keystore.location=keystore_file_path.jks
ssl.keystore.password=keystore_password
ssl.key.password=key_password
如果您可以访问服务器的JMX接口,则开始和结束偏移位于:
kafka.log:type=Log,name=LogStartOffset,topic=TOPICNAME,partition=PARTITIONNUMBER
kafka.log:type=Log,name=LogEndOffset,topic=TOPICNAME,partition=PARTITIONNUMBER
(你需要更换TOPICNAME
和PARTITIONNUMBER
)。请记住,您需要检查给定分区的每个副本,或者您需要找出哪个代理是给定分区的领导者(这可能会随着时间而变化)。
或者,您可以使用Kafka Consumer方法beginningOffsets
和endOffsets
。
我发现最简单的方法是使用Kafdrop REST API /topic/topicName
并指定密钥:"Accept"
/ value:"application/json"
header以获取JSON响应。
它不是java,但可能有用
./bin/kafka-run-class.sh kafka.tools.GetOffsetShell
--broker-list <broker>: <port>
--topic <topic-name> --time -1 --offsets 1
| awk -F ":" '{sum += $3} END {print sum}'
使用https://prestodb.io/docs/current/connector/kafka-tutorial.html
由Facebook提供的超级SQL引擎,连接多个数据源(Cassandra,Kafka,JMX,Redis ......)。
PrestoDB作为带有可选工作服务器的服务器运行(有一个没有额外工作者的独立模式),然后使用一个小的可执行JAR(称为presto CLI)来进行查询。
一旦配置好Presto服务器,就可以使用传统的SQL:
SELECT count(*) FROM TOPIC_NAME;
有时,人们感兴趣的是知道每个分区中的消息数量,例如,在测试自定义分区程序时。随后的步骤已经过测试,可以与Confluent 3.2中的Kafka 0.10.2.1-2一起使用。鉴于Kafka主题,kt
和以下命令行:
$ kafka-run-class kafka.tools.GetOffsetShell \
--broker-list host01:9092,host02:9092,host02:9092 --topic kt
打印示例输出,显示三个分区中的消息计数:
kt:2:6138
kt:1:6123
kt:0:6137
行数可以更多或更少,具体取决于主题的分区数。
由于不再支持ConsumerOffsetChecker
,您可以使用此命令检查主题中的所有消息:
bin/kafka-run-class.sh kafka.admin.ConsumerGroupCommand \
--group my-group \
--bootstrap-server localhost:9092 \
--describe
其中LAG
是主题分区中的消息计数:
您也可以尝试使用kafkacat。这是一个开源项目,可以帮助您从主题和分区中读取消息并将其打印到stdout。这是一个从sample-kafka-topic
主题读取最后10条消息的示例,然后退出:
kafkacat -b localhost:9092 -t sample-kafka-topic -p 0 -o -10 -e
Apache Kafka命令在主题的所有分区上获取未处理的消息:
kafka-run-class kafka.tools.ConsumerOffsetChecker
--topic test --zookeeper localhost:2181
--group test_group
打印:
Group Topic Pid Offset logSize Lag Owner
test_group test 0 11051 11053 2 none
test_group test 1 10810 10812 2 none
test_group test 2 11027 11028 1 none
第6列是未处理的消息。像这样添加它们:
kafka-run-class kafka.tools.ConsumerOffsetChecker
--topic test --zookeeper localhost:2181
--group test_group 2>/dev/null | awk 'NR>1 {sum += $6}
END {print sum}'
awk读取行,跳过标题行并添加第6列,最后打印总和。
打印
5
要获取为主题存储的所有消息,您可以在每个分区的流的开头和结尾搜索使用者并对结果求和
List<TopicPartition> partitions = consumer.partitionsFor(topic).stream()
.map(p -> new TopicPartition(topic, p.partition()))
.collect(Collectors.toList());
consumer.assign(partitions);
consumer.seekToEnd(Collections.emptySet());
Map<TopicPartition, Long> endPartitions = partitions.stream()
.collect(Collectors.toMap(Function.identity(), consumer::position));
consumer.seekToBeginning(Collections.emptySet());
System.out.println(partitions.stream().mapToLong(p -> endPartitions.get(p) - consumer.position(p)).sum());