java实现kafka单机版测试

always_fly

我的系统是centos7（64位）

java环境是：



kafka安装目录：



需要修改config目录下的server.properties

host.name=192.168.3.224（本机ip）

log.dirs=/opt/local/kafka-0.8.1.1-src/logs（日志路径-自定义）



然后是启动：bin/zookeeper-server-start.sh config/zookeeper.properties  &

                  bin/kafka-server-start.sh config/server.properties &

查看是否启动成功，可以查看9092端口和2181端口



创建test主题：bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partiti
ons 1 --topic test

打开生产者：bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

打开消费者：bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginni
ng

在生产者输入内容，消费者就会马上看到

下面是java实现的发送消息和消费消息

java生产者：

1. import java.util.Date;  
   
2. import java.util.Properties;  
   
3. import kafka.javaapi.producer.Producer;  
   
4. import kafka.producer.KeyedMessage;  
   
5. import kafka.producer.ProducerConfig;  
   
6.   
   
7. public class TestProducer {  
   
8.     public static void main(String[] args) {  
   
9.    
   
10.         // 设置配置属性  
    
11.         Properties props = new Properties();  
    
12.         props.put("metadata.broker.list","192.168.3.224:9092");  
    
13.         props.put("serializer.class", "kafka.serializer.StringEncoder");  
    
14.         // key.serializer.class默认为serializer.class  
    
15.         props.put("key.serializer.class", "kafka.serializer.StringEncoder");  
    
16.         // 可选配置，如果不配置，则使用默认的partitioner  
    
17. //        props.put("partitioner.class", "com.catt.kafka.demo.PartitionerDemo");  
    
18.         // 触发acknowledgement机制，否则是fire and forget，可能会引起数据丢失  
    
19.         // 值为0,1,-1,可以参考  
    
20.         // http://kafka.apache.org/08/configuration.html  
    
21.         props.put("request.required.acks", "1");  
    
22.         ProducerConfig config = new ProducerConfig(props);  
    
23.    
    
24.         // 创建producer  
    
25.         Producer<String, String> producer = new Producer<String, String>(config);  
    
26.         // 产生并发送消息  
    
27.         long start=System.currentTimeMillis();  
    
28.         long runtime = new Date().getTime();  
    
29.         String ip = "192.168.3.224" ;   //rnd.nextInt(255);  
    
30.         String msg = runtime + "小张666777" + ip;  
    
31.         //如果topic不存在，则会自动创建，默认replication-factor为1，partitions为0  
    
32.         KeyedMessage<String, String> data = new KeyedMessage<String, String>(  
    
33.                 "test456", ip, msg);  
    
34.         producer.send(data);  
    
35.         System.out.println("耗时:" + (System.currentTimeMillis() - start));  
    
36.         // 关闭producer  
    
37.         producer.close();  
    
38.     }  
    
39. }  
    
40. 
    
41. java消费者：
    
42. [java] view plain copy
    
43. import java.util.HashMap;  
    
44. import java.util.List;  
    
45. import java.util.Map;  
    
46. import java.util.Properties;  
    
47. import kafka.consumer.ConsumerConfig;  
    
48. import kafka.consumer.ConsumerIterator;  
    
49. import kafka.consumer.KafkaStream;  
    
50. import kafka.javaapi.consumer.ConsumerConnector;  
    
51.   
    
52. public class Consumer extends Thread {  
    
53.     private final ConsumerConnector consumer;  
    
54.     private final String topic;  
    
55.     private final String name;  
    
56.   
    
57.     public Consumer(String name, String topic) {  
    
58.         consumer = kafka.consumer.Consumer  
    
59.                 .createJavaConsumerConnector(createConsumerConfig());  
    
60.         this.topic = topic;  
    
61.         this.name = name;  
    
62.     }  
    
63.   
    
64.     private static ConsumerConfig createConsumerConfig() {  
    
65.         Properties props = new Properties();  
    
66.         props.put("zookeeper.connect","192.168.3.224:2181");  
    
67.         props.put("group.id","jd-group");  
    
68.         props.put("zookeeper.session.timeout.ms", "60000");  
    
69.         props.put("zookeeper.sync.time.ms", "2000");  
    
70.         props.put("auto.commit.interval.ms", "1000");  
    
71.         // 每次最少接收的字节数，默认是1  
    
72.         // props.put("fetch.min.bytes", "1024");  
    
73.         // 每次最少等待时间，默认是100  
    
74.         // props.put("fetch.wait.max.ms", "600000");  
    
75.         return new ConsumerConfig(props);  
    
76.     }  
    
77.   
    
78.     public void run() {  
    
79.         Map<String, Integer> topicCountMap = new HashMap<String, Integer>();  
    
80.         topicCountMap.put(topic, new Integer(1));  
    
81.         Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer  
    
82.                 .createMessageStreams(topicCountMap);  
    
83.         KafkaStream<byte[], byte[]> stream = consumerMap.get(topic).get(0);  
    
84.         ConsumerIterator<byte[], byte[]> it = stream.iterator();  
    
85.         while (it.hasNext()) {  
    
86.             System.out.println("************" + name + "    gets    "  
    
87.                     + new String(it.next().message()));  
    
88.         }  
    
89.     }  
    
90. }  
    
91. [java] view plain copy
    
92. public class KafkaConsumerDemo {  
    
93.     public static void main(String[] args) {  
    
94.          Consumer consumerThread1 = new Consumer("Consumer1","test123");  
    
95.   
    
96.          consumerThread1.start();  
    
97.     }  
    
98. }  

复制代码

海海豚

谢谢分享~