<dependency>

    <groupId>org.apache.kafka</groupId>

    <artifactId>kafka-clients</artifactId>

    <version>0.10.2.0</version>

</dependency>

package org.kafka;

import kafka.admin.DeleteTopicCommand;

import kafka.admin.TopicCommand;

/**

 * kafka主题操作

 */

public class TopicDemo {

	/**

	 * 添加主题

	 * linux命令：bin/kafka-topics.sh --create --zookeeper 192.168.2.100:2181 --replication-factor 3 --partitions 1 --topic topictest0416

	 */

	public static void createTopic() {

		String[] options = new String[] {

				"--create",

				"--zookeeper",

				"192.168.2.100:2181",

				"--replication-factor",

				"3",

				"--partitions",

				"1",

				"--topic",

				"topictest0416" };

		TopicCommand.main(options);

	}

	/**

	 * 查询所有主题

	 * linux命令：bin/kafka-topics.sh --list --zookeeper 192.168.2.100:2181

	 */

	public static void queryTopic() {

		String[] options = new String[] {

				"--list",

				"--zookeeper",

				"192.168.2.100:2181" };

		TopicCommand.main(options);

	}

	/**

	 * 查看指定主题的分区及副本状态信息

	 * bin/kafka-topics.sh --describe --zookeeper 192.168.2.100:2181 --topic topictest0416

	 */

	public static void queryTopicByName() {

		String[] options = new String[]{

			    "--describe",

			    "--zookeeper",

			    "192.168.2.100:2181",

			    "--topic",

			    "topictest0416",

			};

		TopicCommand.main(options);

	}

	/**

	 * 修改主题

	 * linux命令：bin/kafka-topics.sh --zookeeper 192.168.2.100:2181 --alter --topic topictest0416 --partitions 3

	 */

	public static void alterTopic() {

		String[] options = new String[]{

			    "--alter",

			    "--zookeeper",

			    "192.168.2.100:2181",

			    "--topic",

			    "topictest0416",

			    "--partitions",

			    "3"

			};

			TopicCommand.main(options);

	}

	/**

	 * 删除主题

	 */

	public static void delTopic() {

		String[] options = new String[] {

				"--zookeeper",

			    "192.168.2.100:2181",

			    "--topic",

			    "topictest0416" };

		DeleteTopicCommand.main(options);

	}

}

package org.kafka;

import java.util.Properties;

import kafka.javaapi.producer.Producer;

import kafka.producer.KeyedMessage;

import kafka.producer.ProducerConfig;

public class ProducerDemo {

	public static void main(String[] args) throws InterruptedException {

		Properties props = new Properties();

		//zookeeper集群列表

		props.put("zk.connect", "hadoop1-1:2181,hadoop1-2:2181,hadoop1-3:2181");

		props.put("metadata.broker.list", "hadoop1-1:9092,hadoop1-2:9092,hadoop1-3:9092");

		//设置消息使用哪个类来序列化

		props.put("serializer.class", "kafka.serializer.StringEncoder");

		ProducerConfig config = new ProducerConfig(props);

		//构造Producer对象

		Producer<String, String> producer = new Producer<String, String>(config);

		// 发送业务消息

		// 读取文件 读取内存数据库

		for (int i = 0; i < 10; i++) {

			Thread.sleep(500);

			KeyedMessage<String, String> km = new KeyedMessage<String, String>("topictest0416", "I am a producer " + i + " hello!");

			producer.send(km);

		}

	}

}

package org.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.KafkaStream;

import kafka.javaapi.consumer.ConsumerConnector;

import kafka.message.MessageAndMetadata;

public class ConsumerDemo {

	private static final String topic = "topictest0416";

	private static final Integer threads = 1;

	public static void main(String[] args) {

		Properties props = new Properties();

		//zookeeper集群列表

		props.put("zookeeper.connect", "hadoop1-1:2181,hadoop1-2:2181,hadoop1-3:2181");

		//消费者组ID

		props.put("group.id", "001");

		//设置读取的偏移量；smallest意思是指向最小的偏移量

		props.put("auto.offset.reset", "smallest");

		//将Properties封装成消费者配置对象

		ConsumerConfig config = new ConsumerConfig(props);

		ConsumerConnector consumer =  Consumer.createJavaConsumerConnector(config);

		Map<String, Integer> topicMap = new HashMap<>();

		//key为消费的topic

		//value为消费的线程数量

		topicMap.put(topic, threads);

		Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicMap);

		List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);

		for (final KafkaStream<byte[], byte[]> kafkaStream : streams) {

			new Thread(new Runnable() {

				@Override

				public void run() {

					for (MessageAndMetadata<byte[], byte[]> mm : kafkaStream) {

						System.out.println(new String(mm.message()));

					}

				}

			}).start();

		}

	}

}