如何检查某些消费者是否使用java连接到Kafka 0.9.0.x？
如何在Kafka上获取连接消费者的列表？作为消费者连接在代理上，是否有像ZkClient/ZkUtils这样的java实用工具在Kafka 0.9.0.x中获得连接消费者的列表？就像我们使用下面的实用程序获取代理列表一样：
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 lib：
<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，它展示了控制台--消费者--6008，但我的编程用户却没有显示出来。使用zkClient.getChildren(ZkUtils.ConsumersPath())，我现在只能查看提到的消费者。
回答 3
Stack Overflow用户
发布于 2016-10-17 22:25:21
不清楚你到底需要什么信息，但我做了一个示例程序，它给出了和kafka相同的信息--消费者组，do描述。
若要使用此代码，请将此依赖项添加到pom中。
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.11</artifactId>
<version>0.9.0.1</version>
</dependency>然后：
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import kafka.admin.AdminClient;
import kafka.coordinator.GroupOverview;
Properties props = new Properties();
props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "kafka-1:9092");
AdminClient adminClient = AdminClient.create(props);
List<GroupOverview> groups = scala.collection.JavaConversions.seqAsJavaList(
adminClient.listAllConsumerGroupsFlattened());
for (GroupOverview group : groups) {
String groupId = group.groupId();
Properties consProps = new Properties();
consProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-1:9092");
consProps.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
consProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
consProps.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
consProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
consProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer consumer = new KafkaConsumer(consProps);
List<AdminClient.ConsumerSummary> groupSummaries = scala.collection.JavaConversions.seqAsJavaList(
adminClient.describeConsumerGroup(groupId));
System.out.println("GROUP, TOPIC, PARTITION, CURRENT OFFSET, LOG END OFFSET, LAG, OWNER");
for (AdminClient.ConsumerSummary summary : groupSummaries) {
String owner = summary.clientId() + "_" + summary.clientHost();
List<TopicPartition> topicPartitions = scala.collection.JavaConversions.seqAsJavaList(
summary.assignment());
for (TopicPartition tp : topicPartitions) {
// Get current offset
long currentOffset = consumer.committed(tp).offset();
// get log end offset
consumer.assign(Arrays.asList(tp));
consumer.seekToEnd();
long logEndOffset = consumer.position(tp);
long lag = logEndOffset - currentOffset;
System.out.println(groupId + ", " + tp.topic() + ", " + tp.partition() + ", " +
currentOffset + ", " + logEndOffset + ", " + lag + ", " + owner);
}
}
}Stack Overflow用户
发布于 2016-10-17 15:44:34
对于0.9.x新消费者，并列出所有活跃的消费者组：
- 查找所有经纪人并向每个经纪人发送"ListGroups“请求，并获取所有组信息；
有关详细信息，请参阅$KAFKA_HOME/bin/kafka-consumer-groups.sh(kafka.admin.ConsumerGroupCommand.KafkaConsumerGroupService.list())
关于0.9.x新消费者和描述某些消费者组的详细信息：
- 查找用户组协调器并向其发送"DescribeGroups“请求，获取所有组成员的信息和分区分配信息；
- 调用KafkaConsumer.committed(TopicPartition分区)获取给定分区的最后提交偏移量。
有关详细信息，请参阅$KAFKA_HOME/bin/kafka-consumer-groups.sh(kafka.admin.ConsumerGroupCommand.KafkaConsumerGroupService.describe())
请注意，旧消费者和新消费者在这方面有完全不同的实现(这两种逻辑都是用kafka.admin.ConsumerGroupCommand.实现的。
Stack Overflow用户
发布于 2016-10-17 11:10:55
这几乎是一样的，但您必须检查ZkUtils.ConsumersPath (= /consumers)。
Zoo门将中的使用者结构是下一个/ each /groupId/ in /consumerId，这样导航您就可以得到每个组的组和使用者。
https://stackoverflow.com/questions/40080807
复制相似问题
腾讯云开发者
