如何获取Kafka上的互联消费者列表?由于消费者是在代理上连接的,在kafka 0.9.0.x中是否有类似zkclient/zkutils这样的java实用程序来获取连接消费者的列表?就像我们使用下面的实用程序获取代理列表一样:
ZkClient zkClient = new ZkClient(endpoint.getZookeeperConnect(), 60000);
if(zkClient!=null){
List<String> brokerIds = zkClient.getChildren(ZkUtils.BrokerIdsPath());
if(CollectionUtils.isNotEmpty(brokerIds) && brokerIds.contains(brokerId)){
logger.debug("Broker:{{}} is connected to Zookeeper.",brokerId);
flag = true;
}
else{
logger.error("ERROR:Broker:{{}} is not connected to Zookeeper.",brokerId);
}
zkClient.close();
}
我正在使用kafka 0.9.0.x和maven的以下java库:
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.11</artifactId>
<version>0.9.0.1</version>
</dependency>
更新时间:
我打开了一个“kafka console consumer.bat”,运行了一次,然后越过cmd提示符。然后转到“zookeeper shell.bat”,然后显示ls/consumers[console-consumer-6008],但是没有显示我编程的consumer。使用 zkClient.getChildren(ZkUtils.ConsumersPath())
我现在只能查看提到的消费者。
3条答案
按热度按时间jgzswidk1#
基本相同,但必须检查zkutils.consumerspath(=/consumers)。
zookeeper中的consumers结构是下一个/consumers/[groupid]/ids/[consumerid],因此导航时可以获得每个组的组和使用者。
l5tcr1uw2#
对于0.9.x新消费者,请列出所有活动消费者组:
找到所有经纪人,向每个经纪人发送“listgroups”请求,获取所有的组信息;
有关详细信息,请参阅
$KAFKA_HOME/bin/kafka-consumer-groups.sh
(kafka.admin.consumergroupcommand.kafkaconsumergroupservice.list())对于0.9.x新消费者,请描述特定消费者群体的详细信息:
找到用户组协调器,向其发送“describegroups”请求,获取所有组成员信息和分区分配信息;
调用kafkaconsumer.committed(topicpartition partition)获取给定分区的最后一个提交的偏移量。
有关详细信息,请参阅
$KAFKA_HOME/bin/kafka-consumer-groups.sh
(kafka.admin.consumergroupcommand.kafkaconsumergroupservice.describe())请注意,旧使用者和新使用者对此有完全不同的实现(这两种逻辑都在kafka.admin.consumergroupcommand中实现)。
nzk0hqpo3#
我不确定你到底需要什么信息,但我做了一个示例程序,它提供了与kafka-consumer-groups.sh相同的信息——描述。
要使用此代码,请将此依赖项添加到pom中。
然后: