kafka入门java例子

1，生产者

import java.util.Properties;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

public class TestProducer {  

        public static void main(String[] args) {
            Properties props = new Properties();
            props.setProperty("metadata.broker.list","10.XX.XX.XX:9092");
            props.setProperty("serializer.class","kafka.serializer.StringEncoder");
            props.put("request.required.acks","1");
            ProducerConfig config = new ProducerConfig(props);
            Producer<String, String> producer = new Producer<String, String>(config);
            KeyedMessage<String, String> data = new KeyedMessage<String, String>("mykafka","test-kafka");
            try {
            	int i =1;
            	while(i < 1000){

            		producer.send(data);
            	}
            } catch (Exception e) {
                e.printStackTrace();
            }
            producer.close();
        }
}

2，消费者

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; 

public class TestConsumer extends Thread{
        private final ConsumerConnector consumer;
        private final String topic;  

        public static void main(String[] args) {
            TestConsumer consumerThread = new TestConsumer("mykafka");
            consumerThread.start();
        }
        public TestConsumer(String topic) {
            consumer =kafka.consumer.Consumer
                    .createJavaConsumerConnector(createConsumerConfig());
            this.topic =topic;
        }  

    private static ConsumerConfig createConsumerConfig() {
        Properties props = new Properties();
        props.put("zookeeper.connect","10.XX.XX.XX:2181,10.XX.XX.XX:2181,10.XX.XX.XX:2181");
        props.put("group.id", "0");
        props.put("zookeeper.session.timeout.ms","10000");
        return new ConsumerConfig(props);
    }  

    public void run(){
        Map<String,Integer> topickMap = new HashMap<String, Integer>();
        topickMap.put(topic, 1);
        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();
        System.out.println("*********Results********");
        while(true){
        	if(it.hasNext()){

        		System.err.println("get data:" +new String(it.next().message()));
        	}
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}  

3，分别启动生产者和消费者，在消费者输出中看到下日志即成功

log4j:WARN No appenders could be found for logger (kafka.utils.VerifiableProperties).
log4j:WARN Please initialize the log4j system properly.
*********Results********
get data:test-kafka

4，启动生产者如果报错如下:

log4j:WARN No appenders could be found for logger (kafka.utils.VerifiableProperties).
log4j:WARN Please initialize the log4j system properly.
kafka.common.FailedToSendMessageException: Failed to send messages after 3 tries.
	at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:90)
	at kafka.producer.Producer.send(Producer.scala:76)
	at kafka.javaapi.producer.Producer.send(Producer.scala:33)
	at ProducerTest.main(TestProducer.java:21)

需要改动config文件夹下的server.properties中的以下两个属性

zookeeper.connect=localhost:2181改成zookeeper.connect=10.0.30.221:2181
以及默认注释掉的
#host.name=localhost改成host.name=10.0.30.221

kafka入门java例子