kafka 0.8.2 消息消費者 consumer

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.hashleaf</groupId>
    <artifactId>kafka</artifactId>
    <version>0.0.1-SNAPSHOT</version>

    <dependencies>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka_2.9.2</artifactId>
            <version>0.8.2.1</version>
        </dependency>
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>1.2.15</version>
            <exclusions>
                <exclusion>
                    <artifactId>jmxtools</artifactId>
                    <groupId>com.sun.jdmk</groupId>
                </exclusion>
                <exclusion>
                    <artifactId>jmxri</artifactId>
                    <groupId>com.sun.jmx</groupId>
                </exclusion>
                <exclusion>
                    <artifactId>jms</artifactId>
                    <groupId>javax.jms</groupId>
                </exclusion>
                <exclusion>
                    <artifactId>mail</artifactId>
                    <groupId>javax.mail</groupId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.11</version>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>1.6</source>
                    <target>1.6</target>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

package com.hashleaf.kafka;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;
import kafka.serializer.Decoder;
import kafka.utils.VerifiableProperties;

/**
 * 自定義消息消費者
 * @author xiaojf [e-mail]
 * @since 2015-7-15 下午11:10:28
 */
public class MyConsumer {
    
    private final ConsumerConnector consumer;
    
    public MyConsumer(){
        Properties originalProps = new Properties();
        
        //zookeeper 配置，通過zk 可以負載均衡的獲取broker
        originalProps.put("zookeeper.connect", "192.168.66.2:2181,192.168.66.3:2181,192.168.66.4:2181");
        
        //group 代表一個消費組
        originalProps.put("group.id", "hashleaf-group");

        //zk連接超時時間
        originalProps.put("zookeeper.session.timeout.ms", "10000");
        //zk同步時間
        originalProps.put("zookeeper.sync.time.ms", "200");
        //自動提交間隔時間
        originalProps.put("auto.commit.interval.ms", "1000");
        //消息日志自動偏移量,防止宕機后數據無法讀取
        originalProps.put("auto.offset.reset", "smallest");
        //序列化類
        originalProps.put("serializer.class", "kafka.serializer.StringEncoder");
        
        //構建consumer connection 對象
        consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(originalProps));
    }
    
    public void consume(){
        //指定需要訂閱的topic
        Map<String ,Integer> topicCountMap = new HashMap<String, Integer>();
        topicCountMap.put(MyProducer.HASHLEAF_KAFKA_TOPIC, new Integer(5));
        
        //指定key的編碼格式
        Decoder<String> keyDecoder = new kafka.serializer.StringDecoder(new VerifiableProperties());
        //指定value的編碼格式
        Decoder<String> valueDecoder = new kafka.serializer.StringDecoder(new VerifiableProperties());
        
        //獲取topic 和 接受到的stream 集合
        Map<String, List<KafkaStream<String, String>>> map = consumer.createMessageStreams(topicCountMap, keyDecoder, valueDecoder);
        
        //根據指定的topic 獲取 stream 集合
        List<KafkaStream<String, String>> kafkaStreams = map.get(MyProducer.HASHLEAF_KAFKA_TOPIC);
        
        ExecutorService executor = Executors.newFixedThreadPool(4);
        
        
        
        //因為是多個 message組成 message set ， 所以要對stream 進行拆解遍歷
        for(final KafkaStream<String, String> kafkaStream : kafkaStreams){
            
            executor.submit(new Runnable() {
                
                @Override
                public void run() {
                  //拆解每個的 stream
                    ConsumerIterator<String, String> iterator = kafkaStream.iterator();
                    
                    while (iterator.hasNext()) {
                        
                        //messageAndMetadata 包括了 message ， topic ， partition等metadata信息
                        MessageAndMetadata<String, String> messageAndMetadata = iterator.next();
                        
                        System.out.println("message : " + messageAndMetadata.message() + "  partition :  " + messageAndMetadata.partition());
                        
                    }
                }
            });
            
            
        }
    }
    
    public static void main(String[] args) {
        new MyConsumer().consume();
    }

}