转载

搭建kafka运行环境

准备环境

系统
CentOS release 6.6 (Final)
下载并解压
wget http://mirror.bit.edu.cn/apache/kafka/0.8.0/kafka_2.10-0.8.0.tgz
tar -zxvf kafka_2.10-0.8.0.tgz

启动

启动 Zookeeper
bin/zookeeper-server-start.sh config/zookeeper.properties &
启动 kafka
bin/kafka-server-start.sh config/server.properties &
创建 topic

我修改了默认的端口
bin/kafka-topics.sh --create --zookeeper localhost:2182 --replication-factor 1 --partitions 1 --topic testweixuan &
查看创建的 topic
bin/kafka-topics.sh --list --zookeeper localhost:2182

查看topic的详细信息
bin/kafka-topics.sh --describe --zookeeper localhost:2182
测试 producer

运行producer并在控制台中输一些消息：
bin/kafka-console-producer.sh --broker-list 192.168.1.116:2182:9092 --topic testweixuan
测试 consumer
bin/kafka-console-consumer.sh --zookeeper 192.168.1.116:2182 --topic testweixuan --from-beginning
注意：需要开两个终端

搭建开发环境

生产者代码

package com.fengtang; import java.util.Properties; import kafka.javaapi.producer.Producer; import kafka.producer.KeyedMessage; import kafka.producer.ProducerConfig; /**  * Create by fengtang  * 2015/10/8 0008  * KafkaDemo_01  */ public class KafkaProducer {  private final Producer<String, String> producer;  public final static String TOPIC = "TEST-TOPIC";  private KafkaProducer() {   Properties props = new Properties();   /**    * 此处配置的是kafka的端口    */   props.put("metadata.broker.list", "192.168.1.116:9092");   /**    *  配置value的序列化类    */   props.put("serializer.class", "kafka.serializer.StringEncoder");   /**    * 配置key的序列化类    */   props.put("key.serializer.class", "kafka.serializer.StringEncoder");   props.put("request.required.acks", "-1");   producer = new Producer<>(new ProducerConfig(props));  }  void produce() {   int messageNo = 1000;   final int COUNT = 10000;   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();  } }

消费者代码

package com.fengtang;  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;  import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Properties;  /**  * Create by fengtang  * 2015/10/8 0008  * KafkaDemo_01  */ public class KafkaConsumer {     private final ConsumerConnector consumer;     private KafkaConsumer() {         Properties props = new Properties();         /**          * zookeeper 配置          */         props.put("zookeeper.connect", "192.168.1.116:2182");          /**          * group 代表一个消费组          */         props.put("group.id", "jd-group");          /**          * zk连接超时          */         props.put("zookeeper.session.timeout.ms", "400000");         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<String, Integer> topicCountMap = new HashMap<String, Integer>();         topicCountMap.put(KafkaProducer.TOPIC, new Integer(1));         StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());         StringDecoder valueDecoder = new StringDecoder(new VerifiableProperties());         Map<String, List<KafkaStream<String, String>>> consumerMap =                  consumer.createMessageStreams(topicCountMap, keyDecoder, valueDecoder);         KafkaStream<String, String> stream = consumerMap.get(KafkaProducer.TOPIC).get(0);         ConsumerIterator<String, String> it = stream.iterator();         while (it.hasNext())             System.out.println(it.next().message());     }      public static void main(String[] args) {         new KafkaConsumer().consume();     } }

pom.xml

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">  <modelVersion>4.0.0</modelVersion>  <groupId>com.fengtang</groupId>  <artifactId>kafkademo</artifactId>  <version>1.0-SNAPSHOT</version>  <packaging>jar</packaging>  <name>kafkademo</name>  <url>http://maven.apache.org</url>  <properties>   <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>  </properties>  <dependencies>   <dependency>    <groupId>org.apache.kafka</groupId>    <artifactId>kafka_2.10</artifactId>    <version>0.8.0</version>   </dependency>   <dependency>    <groupId>log4j</groupId>    <artifactId>log4j</artifactId>    <version>1.2.15</version>    <exclusions>     <exclusion>      <artifactId>jmxtools</artifactId>      <groupId>com.sun.jdmk</groupId>     </exclusion>     <exclusion>      <artifactId>jmxri</artifactId>      <groupId>com.sun.jmx</groupId>     </exclusion>     <exclusion>      <artifactId>jms</artifactId>      <groupId>javax.jms</groupId>     </exclusion>     <exclusion>      <artifactId>mail</artifactId>      <groupId>javax.mail</groupId>     </exclusion>    </exclusions>   </dependency>  </dependencies> </project>