（08）java程序连接kafka示例

　　1、导入kafka所需要的包

　　在服务器上安装kafka程序的时候，解压后就有kafka需要的jar包，如下图所示：

　　2、新建生产者类

package demo;

import java.util.Properties;
import java.util.concurrent.TimeUnit;

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

public class ProducerDemo extends Thread {
    
    //指定具体的topic
    private String topic;
    
    public ProducerDemo(String topic){
        this.topic = topic;
    }
    
    //每隔5秒发送一条消息
    public void run(){
        //创建一个producer的对象
        Producer producer = createProducer();
        //发送消息
        int i = 1;
        while(true){
            String data = "message " + i++;
            //使用produer发送消息
            producer.send(new KeyedMessage(this.topic, data));
            //打印
            System.out.println("发送数据：" + data);
            try {
                TimeUnit.SECONDS.sleep(5);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
    
    //创建Producer的实例
    private Producer createProducer() {
        Properties prop = new Properties();
        //声明zk
        prop.put("zookeeper.connect", "192.168.7.151:2181,192.168.7.152:2181,192.168.7.153:2181");
        prop.put("serializer.class",StringEncoder.class.getName());
        //声明Broker的地址
        prop.put("metadata.broker.list","192.168.7.151:9092,192.168.7.151:9093");
        return new Producer(new ProducerConfig(prop));
    }
    
    public static void main(String[] args) {
        //启动线程发送消息
        new ProducerDemo("mydemo1").start();
    }
}

　　3、新建消费者类

package demo;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;


import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

public class ConsumerDemo extends Thread {

    //指定具体的topic
    private String topic;
    
    public ConsumerDemo(String topic){
        this.topic = topic;
    }
    
    public void run(){
        //构造一个consumer的对象
        ConsumerConnector consumer = createConsumer();
        //构造一个Map对象，代表topic
        //String: topic的名称  Integer: 从这个topic中获取多少条记录
        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        //一次从这个topic中获取一条记录
        topicCountMap.put(this.topic, 1);
        //构造一个messageStream：输入流
        //String: topic的名称 List: 获取的数据
        Map<String, List<KafkaStream<byte[], byte[]>>> messageStreams = consumer.createMessageStreams(topicCountMap);
        //获取每次接受到的具体的数据
        KafkaStream<byte[], byte[]> stream = messageStreams.get(this.topic).get(0);
        ConsumerIterator<byte[], byte[]> iterator = stream.iterator();
        while(iterator.hasNext()){
            String message = new String(iterator.next().message());
            System.out.println("接受数据：" + message);
        }
    }
    
    //创建具体的consumer
    private ConsumerConnector createConsumer() {
        Properties prop = new Properties();
        //指明zk的地址
        prop.put("zookeeper.connect", "192.168.7.151:2181,192.168.7.152:2181,192.168.7.153:2181");
        //指明这个consumer的消费组
        prop.put("group.id", "group1");
        //时间设置的过小可能会连接超时。。。
        prop.put("zookeeper.connection.timeout.ms", "60000");
        return Consumer.createJavaConsumerConnector(new ConsumerConfig(prop));
    }

    public static void main(String[] args) {
        new ConsumerDemo("mydemo1").start();
    }

}

　　运行程序如下：

 　　注意：

　　1、在消费者的类中，时间要设置长一些，否则可能出现连接超时的错误（我就出现了。。。）

　　2、直接关闭生产者和消费者窗口，重新打开消费者窗口，会有重复数据。。。目前还没找到解决办法。。。