分享

kafkaoffsetmonitor监控不到我的group

我安装 完kafkaoffsetmonitor后,启动正常 ,访问页面也可以,就是首页显示的消费 组名不是正确,监控不到我的组名,为什么,显示的是jd-group,不知道这个组名怎么 来的,

已有(1)人评论

跳转到指定楼层
nextuser 发表于 2016-9-30 07:57:15
肯定是楼主配置文件中或则代码中

  //group 代表一个消费组
        props.put("group.id", "jd-group");


下面是一个示例
消费者代码
package com.fengtang;

import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.serializer.StringDecoder;
import kafka.utils.VerifiableProperties;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

/**
* Create by fengtang
* KafkaDemo_01
*/
public class KafkaConsumer {
    private final ConsumerConnector consumer;
    private KafkaConsumer() {
        Properties props = new Properties();
        /**
         * zookeeper 配置
         */
        props.put("zookeeper.connect", "192.168.1.116:2182");

        /**
         * group 代表一个消费组
         */
        props.put("group.id", "jd-group");

        /**
         * zk连接超时
         */
        props.put("zookeeper.session.timeout.ms", "400000");
        props.put("zookeeper.sync.time.ms", "200");
        props.put("auto.commit.interval.ms", "1000");
        props.put("auto.offset.reset", "smallest");
        /**
         * 序列化类
         */
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        ConsumerConfig config = new ConsumerConfig(props);
        consumer = kafka.consumer.Consumer.createJavaConsumerConnector(config);
    }

    void consume() {
        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        topicCountMap.put(KafkaProducer.TOPIC, new Integer(1));
        StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());
        StringDecoder valueDecoder = new StringDecoder(new VerifiableProperties());
        Map<String, List<KafkaStream<String, String>>> consumerMap =
                consumer.createMessageStreams(topicCountMap, keyDecoder, valueDecoder);
        KafkaStream<String, String> stream = consumerMap.get(KafkaProducer.TOPIC).get(0);
        ConsumerIterator<String, String> it = stream.iterator();
        while (it.hasNext())
            System.out.println(it.next().message());
    }

    public static void main(String[] args) {
        new KafkaConsumer().consume();
    }
}

回复

使用道具 举报

您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

关闭

推荐上一条 /2 下一条