kafka java客户端编程

---

kafka_2.10-0.8.1.1

---

maven

<dependencies>
　　<dependency>
　　　　<groupId>org.apache.kafka</groupId>
　　　　<artifactId>kafka_2.10</artifactId>
　　　　<version>0.8.1.1</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>
　　　　</exclusions>
　　</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>
　　<dependency>
　　　　<groupId>junit</groupId>
　　　　<artifactId>junit</artifactId>
　　　　<version>4.11</version>
　　　　<scope>test</scope>
　　</dependency>
</dependencies>

---

producer

package org.admln.kafka.test;

import java.util.Properties;

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

public class Producertest {
     
     public static void main(String[] args) {
         Properties props = new Properties();
         //props.put("zk.connect", "192.168.1.110:2181");
         // serializer.class为消息的序列化类
         props.put("serializer.class", "kafka.serializer.StringEncoder");
         // 配置metadata.broker.list, 为了高可用, 最好配两个broker实例
         props.put("metadata.broker.list", "192.168.1.113:9092");
         // 设置Partition类, 对队列进行合理的划分
         //props.put("partitioner.class", "idoall.testkafka.Partitionertest");
         // ACK机制, 消息发送需要kafka服务端确认
         props.put("request.required.acks", "1");

         props.put("num.partitions", "2");
         ProducerConfig config = new ProducerConfig(props);
         Producer<String, String> producer = new Producer<String, String>(config);
         for (int i = 0; i < 10; i++)
         {
           String msg = "hello" + i;
           producer.send(new KeyedMessage<String, String>("test",msg));
           System.out.println("i:"+i+" msg:"+msg);
         }
       }
}

---

 consumer

 运行consumer一直接收不到消息，还没找到原因

---