這哥們的文章寫的很好，http://my.oschina.net/ielts0909/blog/93190   學習kafka可以讀一讀

我的系統是centos7（64位）

java環境是：

kafka安裝目錄：

需要修改config目錄下的server.properties

host.name=192.168.3.224（本機ip）

log.dirs=/opt/local/kafka-0.8.1.1-src/logs（日誌路徑-自定義）

然後是啟動：bin/zookeeper-server-start.sh config/zookeeper.properties  &

                  bin/kafka-server-start.sh config/server.properties &

檢視是否啟動成功，可以檢視9092埠和2181埠

建立test主題：bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

開啟生產者：bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

開啟消費者：bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

在生產者輸入內容，消費者就會馬上看到

下面是java實現的傳送訊息和消費訊息

java生產者：

import java.util.Date;
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.put("metadata.broker.list","192.168.3.224:9092");
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        // key.serializer.class預設為serializer.class
        props.put("key.serializer.class", "kafka.serializer.StringEncoder");
        // 可選配置，如果不配置，則使用預設的partitioner
//        props.put("partitioner.class", "com.catt.kafka.demo.PartitionerDemo");
        // 觸發acknowledgement機制，否則是fire and forget，可能會引起資料丟失
        // 值為0,1,-1,可以參考
        // http://kafka.apache.org/08/configuration.html
        props.put("request.required.acks", "1");
        ProducerConfig config = new ProducerConfig(props);
 
        // 建立producer
        Producer<String, String> producer = new Producer<String, String>(config);
        // 產生併發送訊息
        long start=System.currentTimeMillis();
        long runtime = new Date().getTime();
        String ip = "192.168.3.224" ;	//rnd.nextInt(255);
        String msg = runtime + "小張666777" + ip;
        //如果topic不存在，則會自動建立，預設replication-factor為1，partitions為0
        KeyedMessage<String, String> data = new KeyedMessage<String, String>(
                "test456", ip, msg);
        producer.send(data);
        System.out.println("耗時:" + (System.currentTimeMillis() - start));
        // 關閉producer
        producer.close();
    }
}
 

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 Consumer extends Thread {
	private final ConsumerConnector consumer;
	private final String topic;
	private final String name;

	public Consumer(String name, String topic) {
		consumer = kafka.consumer.Consumer
				.createJavaConsumerConnector(createConsumerConfig());
		this.topic = topic;
		this.name = name;
	}

	private static ConsumerConfig createConsumerConfig() {
		Properties props = new Properties();
		props.put("zookeeper.connect","192.168.3.224:2181");
		props.put("group.id","jd-group");
		props.put("zookeeper.session.timeout.ms", "60000");
		props.put("zookeeper.sync.time.ms", "2000");
		props.put("auto.commit.interval.ms", "1000");
		// 每次最少接收的位元組數，預設是1
		// props.put("fetch.min.bytes", "1024");
		// 每次最少等待時間，預設是100
		// props.put("fetch.wait.max.ms", "600000");
		return new ConsumerConfig(props);
	}

	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("************" + name + "	gets	"
					+ new String(it.next().message()));
		}
	}
}

public class KafkaConsumerDemo {
	public static void main(String[] args) {
		 Consumer consumerThread1 = new Consumer("Consumer1","test123");

		 consumerThread1.start();
	}
}