Kafka.java 文件源码

public void reconnConsumer(String topicName){

    //停止topic 对应的conn
    ConsumerConnector consumerConn = consumerConnMap.get(topicName);
    consumerConn.commitOffsets(true);
    consumerConn.shutdown();
    consumerConnMap.remove(topicName);

    //停止topic 对应的stream消耗线程
    ExecutorService es = executorMap.get(topicName);
    es.shutdownNow();   
    executorMap.remove(topicName);

    Properties prop = geneConsumerProp();
    ConsumerConnector newConsumerConn = kafka.consumer.Consumer
            .createJavaConsumerConnector(new ConsumerConfig(prop));
    consumerConnMap.put(topicName, newConsumerConn);

    addNewConsumer(topicName, topic.get(topicName));
}