SpoutKafka.java 文件源码

@SuppressWarnings("unchecked")
@Override
public void open(Map conf, TopologyContext context,
        SpoutOutputCollector collector) {
    this._collector = collector;


    Properties props = new Properties();
    props.put("zk.connect", "10.15.62.104:2181");
    props.put("groupid", "group1");

    ConsumerConfig consumerConfig = new ConsumerConfig(props);
    ConsumerConnector consumer = kafka.consumer.Consumer
            .createJavaConsumerConnector(consumerConfig);

    Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
    topicCountMap.put("newtopic", new Integer(1));

    Map<String, List<KafkaMessageStream>> consumerMap = consumer
            .createMessageStreams(topicCountMap);

    KafkaMessageStream stream = consumerMap.get("newtopic").get(0);

    this.it = stream.iterator();        
}