./kafka-topics.sh --zookeeper node3:2181,node4:2181,node5:2181  --create --topic topic2017 --partitions 3 --replication-factor 3

./kafka-console-producer.sh  --topic  topic2017  --broker-list node1:9092,node2:9092,node3:9092

./kafka-console-consumer.sh --zookeeper node3:2181,node4:2181,node5:2181 --topic topic2017

./kafka-topics.sh  --list --zookeeper node3:2181,node4:2181,node5:2181

./kafka-topics.sh --describe --zookeeper node3:2181,node4:2181,node5:2181  --topic topic2017

ls /brokers/topics/

ls /consumers

JAVA连接Kafka

KafkaProducer

package com.kafka;

import java.util.Properties;

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

/**
 * description： 
 *  step1 : 创建存放配置信息的properties
 *  step2 : 将properties封装到ProducerConfig中
 *  step3 : 创建producer对象
 *  step4 : 发送数据流
 *
 * 执行过程：
 * 1、创建一个topic
 * kafka-topic.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 3 --topic xxxx
 * 2、运行本类中的代码
 * 3、查看message
 * kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic xxxx
 * kafka
 */
public class KafkaProducer {

    public static void main(String[] args) throws InterruptedException {
    	//step1 : 创建存放配置信息的properties
    	Properties props = new Properties();
    	/**
         * 指定producer连接的broker列表
         */
    	props.put("metadata.broker.list", "node01:9092,node02:9092,node03:9092");
    	
		/**
		 * ack机制
		 * 0 which means that the producer never waits for an acknowledgement from the broker 
		 * 1 which means that the producer gets an acknowledgement after the leader replica has received the data 
		 * -1 The producer gets an acknowledgement after all in-sync replicas have received the data
		 */
    	props.put("request.required.acks", "1");
    	//消息发送的同步与异步
    	props.put("producer.type", "sync");
    	 /**
         * 指定message的序列化方法，用户可以通过实现kafka.serializer.Encoder接口自定义该类，kafka.serializer.DefaultEncoder
         * 默认情况下message的key和value都用相同的序列化，但是可以使用"key.serializer.class"指定key的序列化
         */
    	props.put("serializer.class", "kafka.serializer.StringEncoder");
    	// 设置自定义的partition，当topic有多个partition时如何对message进行分区
        // props.put("partitioner.class", "cn.com.dimensoft.kafka.SamplePartition");
    	
    	 // step2 : 将properties封装到ProducerConfig中
    	ProducerConfig config = new ProducerConfig(props);
    	
    	// step3 : 创建producer对象,指定泛型 Producer的两个泛型，第一个指定Key的类型，第二个指定value的类型
    	Producer<String,String> producer = new Producer<String,String>(config);
    	
    	//发送数据流
    	for (int i = 150; i < 200 ; i++) {
	    	 /**
	         * 第一个泛型指定用于分区的key的类型，第二个泛型指message的类型
	         * topic只能为String类型
	         */
			producer.send(new KeyedMessage<String, String>("dengchao" , "我是第 "+ i +" 号。"));
			Thread.sleep(100);
		}
    	
    	producer.close();
    	
    }

}

KafkaProducer

package com.kafka;

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;
import kafka.message.MessageAndMetadata;
import kafka.serializer.StringDecoder;
import kafka.utils.VerifiableProperties;


/**
 * description： 
 *  step1 : 创建存放配置信息的properties
 *  step2 : 将properties封装到ConsumerConfig中
 *  step3 : 调用Consumer的静态方法创建ConsumerConnector
 *  step4 : 根据创建好的ConsumerConnector对象创建MessageStreams集合
 *  step5 : 根据具体的topic名称得到数据流KafkaStream
 *  step6 : 调用KafkaStream的iterator拿到ConsumerIterator对应，然后就可以迭代获得producer发送过来的消息了
 */
public class KafkaConsumer {

	private ConsumerConnector connector ;
	
	public KafkaConsumer() {
		Properties props = new Properties();
		
		props.put("zookeeper.connect","node01:2181,node02:2181,node03:2181");
		//设置消费组id  group设置在 /config/consumer.properties
		props.put("group.id", "test-consumer-group");
		
		//下面这2个参数需要设置，否则consumer每次启动都会从头开始读取数据
		
		//自动提交到 zk  注意offset信息并不是每消费一次消息就向zk提交一次,而是现在本地保存(内存),并定期提交
		//可以设置为false  自己手动调用consumer中的commit 方法（其实是讲message中存在的offset信息提交到zk上进行存储）
		props.put("auto.commit.enable", "true");
		
		//自动提交的时间间隔,默认为1分钟  这里是1秒
		props.put("auto.commit.interval.ms", "1000");
		
		//在Consumer在ZooKeeper中发现没有初始的offset时或者发现offset不在范围呢，该怎么做：
		//smallest : 自动把offset设为最小的offset。
		//largest : 自动把offset设为最大的offset。
		//anything else: 抛出异常
		props.put("auto.offset.reset", "smallest");
		
		//request.required.acks=0 时的问题
		//问题：当consumer消费失败后，会导致消息丢失；改进：每次consumer消费后，给broker ack，若broker在超时时间未收到ack，则重发此消息。
		//问题:1.当消费成功，但未ack时，会导致消费2次 
		//	   2. now the broker must keep multiple states about every single message  
		//	   3.当broker是多台机器时，则状态之间需要同步
		props.put("request.required.acks", 1);
		
		//指定序列化处理类,默认为kafka.serializer.DefaultEncoder  
        props.put("serializer.class", "kafka.serializer.StringEncoder");  
		
		// step2 : 将properties封装到ConsumerConfig中
		ConsumerConfig config = new ConsumerConfig(props);
		
		// step3 : 调用Consumer的静态方法创建ConsumerConnector
		connector = Consumer.createJavaConsumerConnector(config);
	}
	
	
	public static void main(String[] args) {
		try {
			new KafkaConsumer().consumer();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
	
	public void consumer() throws InterruptedException{
		// step4 : 根据创建好的ConsumerConnector对象创建MessageStreams集合
		Map<String,Integer> topicCountMap = new HashMap<String,Integer>();
		// 描述读取哪几个topic，该topic需要几个线程读
		topicCountMap.put("dengchao", 1);
		
		StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());
		StringDecoder valueDecoder = new StringDecoder(new VerifiableProperties());
		
		// 根据填充好的map获得streams集合
        // a map of (topic, list of KafkaStream) pairs
		Map<String, List<KafkaStream<String, String>>> streams  = connector.createMessageStreams(topicCountMap,keyDecoder,valueDecoder);
		
		// step5 : 根据具体的topic名称得到数据流KafkaStrea
		// 每个线程对应于一个KafkaStream
		KafkaStream<String, String> kafkaStream = streams.get("dengchao").get(0);
		
		// step6 : 调用KafkaStream的iterator拿到ConsumerIterator
        // 然后就可以迭代获得producer发送过来的消息了
		ConsumerIterator<String, String> iterator = kafkaStream.iterator();
		
		MessageAndMetadata<String, String> mm = null;
		while( iterator.hasNext() ){
			 mm = iterator.next();
			 System.out.println("partition : " + mm.partition() + " , message : " + mm.message());
			 Thread.sleep(50);
		}
		
	}
	
}