本部署使用的版本为kafka_2.8.0-0.8.0。 

参考了http://blog.csdn.net/itleochen/article/details/17451455这篇博文；  并根据官网介绍http://kafka.apache.org/documentation.html#quickstart完成。  废话少说，直接上步骤  1.下载kafka_2.8.0-0.8.0.tar.gz  https://archive.apache.org/dist/kafka/0.8.0/kafka_2.8.0-0.8.0.tar.gz  2.解压缩  tar -vxf kafka_2.8.0-0.8.0.tar.gz  3.修改配置文件  修改conf/server.properties  host.name=192.168.110.129（修改为主机ip,不然服务器返回给客户端的是主机的hostname,客户端并不一定能够识别）  修改conf/zookeeper.properties 属性文件  dataDir=/usr/local/tmp/zookeeper   (zookeeper临时数据文件)  4.启动zookeeper和kafka  cd bin  启动zookeeper  ./zookeeper-server-start.sh ../config/zookeeper.properties & (&推出命令行，服务守护执行)  启动kafka  ./kafka-server-start.sh ../config/server.properties &  5.验证是否成功  *创建主题  ./kafka-create-topic.sh --partition 1 --replica 1 --zookeeper localhost:2181 --topic test  检查是否创建主题成功  ./kafka-list-topic.sh --zookeeper localhost:2181  *启动produce  ./bin/kafka-console-producer.sh --broker-list 192.168.110.129:9092  --topic test  *启动consumer  ./kafka-console-consumer.sh --zookeeper localhost:2181 --topic test  6.关闭kafka和zookeeper  ./kafka-server-stop.sh ../config/server.properties  ./zookeeper-server-stop.sh  心得总结：  1.produce启动的时候参数使用的是kafka的端口而consumer启动的时候使用的是zookeeper的端口；  2.必须先创建topic才能使用；  3.topic本质是以文件的形式储存在zookeeper上的。

 

消费者

package com.kafka;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;import kafka.serializer.StringDecoder;import kafka.utils.VerifiableProperties;public class KafkaConsumer{	private final ConsumerConnector consumer;	private KafkaConsumer()	{		Properties props = new Properties();		// zookeeper 配置		props.put( "zookeeper.connect", "192.168.110.129:2181" );		// group 代表一个消费组		props.put( "group.id", "jd-group" );		// zk连接超时		props.put( "zookeeper.session.timeout.ms", "4000" );		props.put( "zookeeper.sync.time.ms", "200" );		props.put( "auto.commit.interval.ms", "1000" );		props.put( "auto.offset.reset", "smallest" );		// 序列化类		props.put( "serializer.class", "kafka.serializer.StringEncoder" );		ConsumerConfig config = new ConsumerConfig( props );		consumer = kafka.consumer.Consumer.createJavaConsumerConnector( config );	}	void consume()	{		Map
    
      topicCountMap = new HashMap
     
      ();		topicCountMap.put( KafkaProducer.TOPIC, new Integer( 1 ) );		StringDecoder keyDecoder = new StringDecoder( new VerifiableProperties() );		StringDecoder valueDecoder = new StringDecoder( new VerifiableProperties() );		Map
      
       >> consumerMap = consumer.createMessageStreams( topicCountMap, keyDecoder, valueDecoder );		KafkaStream
       
         stream = consumerMap.get( KafkaProducer.TOPIC ).get( 0 );		ConsumerIterator
        
          it = stream.iterator();		while (it.hasNext())			System.out.println( it.next().message() );	}	public static void main(String[] args)	{		new KafkaConsumer().consume();	}}
        
       
      
     
    

生产者

package com.kafka;import java.util.Properties;import kafka.javaapi.producer.Producer;import kafka.producer.KeyedMessage;import kafka.producer.ProducerConfig;/** * Hello world! * */public class KafkaProducer{	private final Producer
    
      producer;	public final static String TOPIC = "TEST-TOPIC";	private KafkaProducer()	{		Properties props = new Properties();		// 此处配置的是kafka的端口		props.put( "metadata.broker.list", "192.168.110.129:9092" );		// 配置value的序列化类		props.put( "serializer.class", "kafka.serializer.StringEncoder" );		// 配置key的序列化类		props.put( "key.serializer.class", "kafka.serializer.StringEncoder" );		// request.required.acks		// 0, which means that the producer never waits for an acknowledgement from the broker (the same		// behavior as 0.7). This option provides the lowest latency but the weakest durability guarantees		// (some data will be lost when a server fails).		// 1, which means that the producer gets an acknowledgement after the leader replica has received the		// data. This option provides better durability as the client waits until the server acknowledges the		// request as successful (only messages that were written to the now-dead leader but not yet		// replicated will be lost).		// -1, which means that the producer gets an acknowledgement after all in-sync replicas have received		// the data. This option provides the best durability, we guarantee that no messages will be lost as		// long as at least one in sync replica remains.		props.put( "request.required.acks", "-1" );		producer = new Producer
     
      ( new ProducerConfig( props ) );	}	void produce()	{		int messageNo = 1000;		final int COUNT = 2000;		while (messageNo < COUNT)		{			String key = String.valueOf( messageNo );			String data = "hello kafka message " + key;			producer.send( new KeyedMessage
      
       ( TOPIC, key, data ) );			// System.out.println( data );			messageNo++;		}	}	public static void main(String[] args)	{		new KafkaProducer().produce();	}}