（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、直接關閉生產者和消費者窗口，重新打開消費者窗口，會有重復數據。。。目前還沒找到解決辦法。。。