1、 請一定要修改 hosts 檔案，不能用 ip 地址，必須域名：
· 檔案路徑：C:\Windows\System32\drivers\etc\hosts；

2、 建立一個新的 Java 專案，將 Kafka 的開發包配置到專案的 CLASSPATH 之中：

3、 建立一個新的生產者用於訊息的傳送；

package cn.test.demo;
import java.util.Properties;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import
 
 kafka.serializer.StringEncoder;
public class SendMessage {
    public static void main(String[] args) throws Exception {
        String topic = "test"; // 定義要操作的主題
            Properties pro = new Properties(); // 定義相應的屬性儲存
            pro.setProperty("zookeeper.connect", "192.168.122.206:2181");
            pro.setProperty("metadata.broker.list"
 
, "kafka-server-alone:9092");
            pro.setProperty("serializer.class", StringEncoder.class.getName());
            Producer<Integer, String> prod = new Producer<Integer, String>(new ProducerConfig(pro));
            prod.send(new KeyedMessage<Integer, String>(topic, "test"));
    }
}   
 

4、 建立消費者程式

package cn.test.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 ReceiveMessage {
    public static void main(String[] args) throws Exception {
        String topic = "test"; // 定義要操作的主題
        Properties pro = new Properties(); // 定義相應的屬性儲存
        pro.setProperty("zookeeper.connect", "192.168.122.206:2181");
        pro.setProperty("metadata.broker.list", "kafka-server-alone:9092");
        pro.setProperty("group.id", "group1"); 
        ConsumerConnector consumer = Consumer.createJavaConsumerConnector(new   ConsumerConfig(pro));
        // 需要定義一個主題的對映的儲存集合
        Map<String,Integer> topicMap = new HashMap<String,Integer>() ;
        topicMap.put(topic, 1) ; // 設定要讀取資料的主題
        Map<String, List<KafkaStream<byte[], byte[]>>> messageStreams = 
        consumer.createMessageStreams(topicMap) ;
        // 現在只有一個主題，所以此處只接收第一個主題的資料即可
        // 第一個主題
        KafkaStream<byte[], byte[]> stream = messageStreams.get(topic).get(0) ; 
        ConsumerIterator<byte[], byte[]> iter = stream.iterator() ;
        while(iter.hasNext()) {
            String msg = new String(iter.next().message()) ;
            System.out.println("接收到訊息：" + msg);
        }
    }
}