package com.rta.kafka;

import java.text.SimpleDateFormat;
import java.util.Date;
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 backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichSpout;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
 
public class KafkaSpoutTest implements IRichSpout {
    private SpoutOutputCollector collector;
    private ConsumerConnector consumer;
    private String topic;
 
    public KafkaSpoutTest() {}
     
    public KafkaSpoutTest(String topic) {
        this.topic = topic;
    }
 
    public void nextTuple() {    }
 
    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
        this.collector = collector;
    }
 
    public void ack(Object msgId) {    }
 
    public void activate() {         
        consumer =kafka.consumer.Consumer.createJavaConsumerConnector(createConsumerConfig()); 
        Map<String,Integer> topickMap = new HashMap<String, Integer>();  
        topickMap.put(topic, 1);  
 
        System.out.println("*********Results********topic:"+topic);  
 
        Map<String, List<KafkaStream<byte[],byte[]>>>  streamMap=consumer.createMessageStreams(topickMap);  
        KafkaStream<byte[],byte[]>stream = streamMap.get(topic).get(0);  
        ConsumerIterator<byte[],byte[]> it =stream.iterator();   
        while(it.hasNext()){  
             String value =new String(it.next().message());
             SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd日 HH:mm:ss SSS");
             Date curDate = new Date(System.currentTimeMillis());//获取当前时间       
             String str = formatter.format(curDate);   
             System.out.println("storm接收到来自kafka的消息------->" + value);
             collector.emit(new Values(value,1,str), value);
        }  
    }
     
    private static ConsumerConfig createConsumerConfig() {  
        Properties props = new Properties();  
        // 设置zookeeper的地址
        props.put("zookeeper.connect","localhost:2181");
        // 设置group id
        props.put("group.id", "1");  
        // kafka的group 消费记录是保存在zookeeper上的, 但这个信息在zookeeper上不是实时更新的, 需要有个间隔时间更新
        props.put("auto.commit.interval.ms", "1000");
        props.put("zookeeper.session.timeout.ms","10000");  
        return new ConsumerConfig(props);  
    }  
 
    public void close() {    }
 
    public void deactivate() {    }
 
    public void fail(Object msgId) {    }
 
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("word","id","time"));
    }
 
    public Map<String, Object> getComponentConfiguration() {
        System.out.println("getComponentConfiguration被调用");
        // bin/kafka-topics.sh --list --zookeeper localhost:2181
        topic="test";
        return null;
    }
}
@@ -1912,24 +1912,16 @@
 ());
-//%E8%8E%B7%E5%8F%96%E5%BD%93%E5%89%8D%E6%97%B6%E9%97%B4
        %0A
     
@@ -1908,32 +1908,32 @@
 llis());       %0A
+
              Str
@@ -2026,20 +2026,16 @@
 ka%E7%9A%84%E6%B6%88%E6%81%AF---
-----
 %3E%22 + val
@@ -2234,35 +2234,8 @@
 (); 
- %0A        // %E8%AE%BE%E7%BD%AEzookeeper%E7%9A%84%E5%9C%B0%E5%9D%80
 %0A   
@@ -2292,30 +2292,8 @@
 %22);%0A
-        // %E8%AE%BE%E7%BD%AEgroup id%0A
     
@@ -2337,16 +2337,32 @@
      // 
+%E9%97%B4%E9%9A%94%E6%9B%B4%E6%96%B0zookeeper%E4%B8%8A%E5%85%B3%E4%BA%8E
 kafka%E7%9A%84gr
@@ -2368,63 +2368,16 @@
 roup
- 
 %E6%B6%88%E8%B4%B9%E8%AE%B0%E5%BD%95
-%E6%98%AF%E4%BF%9D%E5%AD%98%E5%9C%A8zookeeper%E4%B8%8A%E7%9A%84, %E4%BD%86%E8%BF%99%E4%B8%AA%E4%BF%A1%E6%81%AF%E5%9C%A8zookeeper%E4%B8%8A%E4%B8%8D%E6%98%AF%E5%AE%9E%E6%97%B6%E6%9B%B4%E6%96%B0%E7%9A%84, %E9%9C%80%E8%A6%81%E6%9C%89%E4%B8%AA%E9%97%B4%E9%9A%94%E6%97%B6%E9%97%B4
+%EF%BC%8C%E9%9D%9E%E5%AE%9E%E6%97%B6
 %E6%9B%B4%E6%96%B0%0A 
@@ -3030,8 +3030,9 @@
 %0A    %7D%0A%7D
+%0A

Update KafkaSpoutTest.java