zoukankan      html  css  js  c++  java
  • kafka java api消费者

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

    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;

    public class KafkaConsumer {

    private final ConsumerConnector consumer;

    private KafkaConsumer() {
    Properties props = new Properties();

    // zookeeper 配置
    props.put("zookeeper.connect", "192.168.170.185:2181");

    // 消费者所在组
    props.put("group.id", "testgroup");

    // zk连接超时
    props.put("zookeeper.session.timeout.ms", "4000");
    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();

    int messageCount = 0;
    while (it.hasNext()){
    System.out.println(it.next().message());
    messageCount++;
    if(messageCount == 100){
    System.out.println("Consumer端一共消费了" + messageCount + "条消息!");
    }
    }
    }

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

  • 相关阅读:
    javascript 将中文符号转换成英文符号
    js 函数定义的2种方式
    javascript 面向对象编程(工厂模式、构造函数模式、原型模式)
    10.2.0.4 to 10.2.0.5 Installation of Patch Set Release (Windows)
    流接口驱动程序的工作原理和开发步骤嵌入式
    android开发面试题
    ORACLE-014:oracle中查看DBLinkpassword
    Java相关知识(一)
    Android 中模仿 Twitter 实现 Toolbar Indicator
    使用excel进行数据挖掘(3)----类别检測
  • 原文地址:https://www.cnblogs.com/wangjing666/p/6860751.html
Copyright © 2011-2022 走看看