kafka 소비자 실례 - 간결하고 사용 가능

package com.z.kafka;

import java.util.Arrays;
import java.util.Collections;
import java.util.Iterator;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;

import com.zjb.PropertiesUtil;

public class ConsumerUtil {
	
	private static final String GROUP = "testGrp";
	private static final long TIMEOUT_MS = 100;
	private static String BROKER_LIST = "127.0.0.1:9092";
    private static String DECODER = "org.apache.kafka.common.serialization.StringDeserializer";

	private KafkaConsumer initConsumer(String grpId) {
		Properties pro = new Properties();
		
		pro.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKER_LIST);
		pro.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, "client_test2");
		pro.setProperty(ConsumerConfig.GROUP_ID_CONFIG, grpId);
		pro.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
		pro.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
		pro.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, DECODER );
		pro.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, DECODER );
		
		KafkaConsumer consumer = new KafkaConsumer(pro);
		return consumer;
	}

	public KafkaConsumer getConsumer(String topic, String group){
		
		String grp = GROUP;
		if(group != null){
			grp = group;
		}
		
		return initConsumer(grp);
	}
	
	public static void main(String[] args) {
		
		String topic = "test_topic";
		String group = "test_grp";
		
		System.out.println("topic:{"+topic+"},group:{"+GROUP+"}");
		KafkaConsumer consumer = new ConsumerUtil().getConsumer(topic, group);
		consumer.subscribe(Arrays.asList(topic));
	     while (true) {
	         ConsumerRecords records = consumer.poll(100);
	         for (ConsumerRecord record : records){
	        	System.out.println(record.value());
	         }
	     }
	}
	
}