---

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

?

1 package org.admln.kafka.test; 2 3 import java.util.Properties; 4 5 import kafka.javaapi.producer.Producer; 6 import kafka.producer.KeyedMessage; 7 import kafka.producer.ProducerConfig; 8 9 public class Producertest { 10 11 public static void main(String[] args) { 12 Properties props = new Properties(); 13 //props.put("zk.connect", "192.168.1.110:2181"); 14 // serializer.class為消息的序列化類 15 props.put("serializer.class", "kafka.serializer.StringEncoder"); 16 // 配置metadata.broker.list, 為了高可用, 最好配兩個broker實例 17 props.put("metadata.broker.list", "192.168.1.113:9092"); 18 // 設置Partition類, 對隊列進行合理的劃分 19 //props.put("partitioner.class", "idoall.testkafka.Partitionertest"); 20 // ACK機制, 消息發送需要kafka服務端確認 21 props.put("request.required.acks", "1"); 22 23 props.put("num.partitions", "2"); 24 ProducerConfig config = new ProducerConfig(props); 25 Producer<String, String> producer = new Producer<String, String>(config); 26 for (int i = 0; i < 10; i++) 27 { 28 String msg = "hello" + i; 29 producer.send(new KeyedMessage<String, String>("test",msg)); 30 System.out.println("i:"+i+" msg:"+msg); 31 } 32 } 33 }

---

?consumer

?運行consumer一直接收不到消息，還沒找到原因

---

?

轉載于:https://www.cnblogs.com/admln/p/kafka-JavaClient-program.html

總結

以上是生活随笔為你收集整理的kafka java客户端编程的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。