Apache kafka之旅——java 发送接受消息

一、新建maven项目，引用apache给定的pom文件，如下

<dependency>
 <groupId>org.apache.kafka</groupId>
 <artifactId>kafka_2.9.2</artifactId>
 <version>0.8.1.1</version>
</dependency>

二新建发送消息类KafkaProducer.java代码如下:

import java.util.Properties;  

import kafka.producer.KeyedMessage;  

import kafka.producer.ProducerConfig; 

public class KafkaProducer extends Thread{
private final kafka.javaapi.producer.Producer<Integer, String> producer;  

    private final String topic;  

    private final Properties props = new Properties();  

  

    public KafkaProducer(String topic)  

    {  

        props.put("serializer.class", "kafka.serializer.StringEncoder");  

        props.put("metadata.broker.list", "192.168.30.229:9092");  

        producer = new kafka.javaapi.producer.Producer<Integer, String>(new ProducerConfig(props));  

        this.topic = topic;  

    }  

  

    @Override  

    public void run() {  

        int messageNo = 1;  

        while (true)  

        {  

            String messageStr = new String("Message_" + messageNo);  

            System.out.println("Send:" + messageStr);  

            producer.send(new KeyedMessage<Integer, String>(topic, messageStr));  

            messageNo++;  

            try {  

                sleep(3000);  

            } catch (InterruptedException e) {  

                e.printStackTrace();  

            }  

        }  

    }  

}

三、新建接受消息类KafkaConsumer.java

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 KafkaConsumer extends Thread{
private final ConsumerConnector consumer;  

    private final String topic;  

  

    public KafkaConsumer(String topic)  

    {  

        consumer = kafka.consumer.Consumer.createJavaConsumerConnector(  

                createConsumerConfig());  

        this.topic = topic;  

    }  

  

    private static ConsumerConfig createConsumerConfig()  

    {  

        Properties props = new Properties();  

        props.put("zookeeper.connect", "192.168.30.229:2181");  

        props.put("group.id", "group1");  

        props.put("zookeeper.session.timeout.ms", "1000000");  

        props.put("zookeeper.sync.time.ms", "200");  

        props.put("auto.commit.interval.ms", "1000");  

        return new ConsumerConfig(props);  

    }  

  

    @Override  

    public void run() {  

        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();  

        topicCountMap.put(topic, new Integer(1));  

        Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);  

        KafkaStream<byte[], byte[]> stream = consumerMap.get(topic).get(0);  

        ConsumerIterator<byte[], byte[]> it = stream.iterator();  

        while (it.hasNext()) {  

            System.out.println("receive：" + new String(it.next().message()));  

            try {  

                sleep(3000);  

            } catch (InterruptedException e) {  

                e.printStackTrace();  

            }  

        }  

    }  

}

四、新建KafkaConsumerProducerDemo，代码：

public class KafkaConsumerProducerDemo {
public static void main(String[] args)  

    {  

        KafkaProducer producerThread = new KafkaProducer("test");  

        producerThread.start();  

  

        KafkaConsumer consumerThread = new KafkaConsumer("test");  

        consumerThread.start();  

    }  

}

以上的ip地址定要改成你的kafka服务器的ip

端口号2181 ，9092 一定要 保证linux上对外开放，或者直接把防火墙关闭

运行第四步就可以实现了