Spark Tips4: Kafka的Consumer Group及其在Spark Streaming中的“异动”(更新)

public class SFQConsumerGroup {
    private final ConsumerConnector consumer;
    private final String topic;
    private  ExecutorService executor;

    public SFQConsumerGroup(String a_zookeeper, String a_groupId, String a_topic) {
        consumer = kafka.consumer.Consumer.createJavaConsumerConnector(
                createConsumerConfig(a_zookeeper, a_groupId));
        this.topic = a_topic;
    }

    public void run(int a_numThreads) {
        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        topicCountMap.put(topic, new Integer(a_numThreads));
        Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
        List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);

        // now launch all the threads
        executor = Executors.newFixedThreadPool(a_numThreads);

        // now create an object to consume the messages
        int threadNumber = 0;
        for (final KafkaStream stream : streams) {
            executor.execute(new ConsumerProcessor(stream, threadNumber));
            threadNumber++;
        }

        executor.shutdown();
        while (!executor.isTerminated()) {

        }
    }

    private static ConsumerConfig createConsumerConfig(String a_zookeeper, String a_groupId) {
        Properties props = new Properties();
        props.put("zookeeper.connect", a_zookeeper);
        props.put("group.id", a_groupId);
        props.put("auto.offset.reset", "smallest");
        props.put("auto.commit.enable", "false");        
        props.put("zookeeper.session.timeout.ms", "400");
        props.put("zookeeper.sync.time.ms", "200");

        return new ConsumerConfig(props);
    }

    public static void main(String[] args) {
        if (args.length < 3) {
            System.err.println("You need to input broker and topicName as arguments. E.g: localhost:2181 newtest1 jull, or: localhost:2181 newtest1 jull 2");
        }

        String zooKeeper = args[0];
        String topic = args[1];
        String groupId = args[2];

        int threadNum = 2;

        if (args.length  > 3) {
          threadNum = Integer.parseInt(args[3]);
        }

        SFQConsumerGroup example = new SFQConsumerGroup(zooKeeper, groupId, topic);
        example.run(threadNum);

        try {
            Thread.sleep(1 * 1000);
        } catch (InterruptedException ie) {

        }        
    }
}

class ConsumerProcessor implements Runnable {
    private KafkaStream m_stream;
    private int m_threadNumber;

    public ConsumerProcessor(KafkaStream a_stream, int a_threadNumber) {
        m_threadNumber = a_threadNumber;
        m_stream = a_stream;
    }

    public void run() {
        ConsumerIterator<byte[], byte[]> it = m_stream.iterator();
        while (it.hasNext())
            System.out.println("Thread " + m_threadNumber + ": " + new String(it.next().message()));
        System.out.println("Shutting down Thread: " + m_threadNumber);
    }
}

... ...

    HashMap<String, String> kafkaParams = new HashMap<String, String>();
    kafkaParams.put("metadata.broker.list", brokerList);
    kafkaParams.put("group.id", groupId);

    JavaPairInputDStream<String, String> messages = KafkaUtils.createDirectStream(
                jssc,
                String.class,
                String.class,
                StringDecoder.class,
                StringDecoder.class,
                kafkaParams,
                topicsSet
        );

    messages.foreachRDD(new Function<JavaPairRDD<String, String>, Void>() {

        @Override
        public Void call(JavaPairRDD<String, String> rdd) throws Exception {
            long msgNum = rdd.count();
            System.out.println("There are " + msgNum + " messages read from Kafka.");

        ... ...

        return null;}});
... ...

Map<String,Integer> topicMap=new HashMap<String,Integer>();
for (int i  = 0; i < topicNames.length; i ++) {
    topicMap.put(topicNames[i], partitionNum[i]);
}

JavaPairReceiverInputDStream<String, String> messages = KafkaUtils.createStream(
    jssc, 
    zkQuorum, 
    groupId, 
    topicMap,
    StorageLevel.MEMORY_ONLY());    

messages.foreachRDD(new Function<JavaPairRDD<String, String>, Void>() {
    @Override
    public Void call(JavaPairRDD<String, String> rdd) throws Exception {                         long msgNum = rdd.count();

        System.out.println("There are " + msgNum + " messages read from Kafka.");
    
         ... ...

         return null;
    }
});