Kafka简单客户端编程实例

时间:2022-09-03 18:18:30

今天,我们给大家带来一篇如何利用kafka的api进行客户端编程的文章,这篇文章很简单,就是利用kafka的api创建一个生产者和消费者,生产者不断向kafka写入消息,消费者则不断消费kafka的消息。下面是具体的实例代码。

一、创建配置类config

这个类很简单,只是存放了两个常量,一个是话题topic,一个是线程数threads

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
package com.lya.kafka;
 
/**
 * 配置项
 * @author liuyazhuang
 *
 */
public class config {
  
 /**
  * 话题
  */
 public static final string topic = "wordcount";
 /**
  * 线程数
  */
 public static final integer threads = 1;
}

二、编程生产者类producerdemo

这个类的主要作用就是向kafka写入相应的消息,并且将消息写入wordcount话题。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
package com.lya.kafka;
 
import java.util.properties;
 
import kafka.javaapi.producer.producer;
import kafka.producer.keyedmessage;
import kafka.producer.producerconfig;
 
/**
 * 生产者实例
 * @author liuyazhuang
 *
 */
public class producerdemo {
 public static void main(string[] args) throws exception {
  properties props = new properties();
  props.put("zk.connect", "192.168.209.121:2181");
  props.put("metadata.broker.list","192.168.209.121:9092");
  props.put("serializer.class", "kafka.serializer.stringencoder");
  props.put("zk.connectiontimeout.ms", "15000");
  producerconfig config = new producerconfig(props);
  producer<string, string> producer = new producer<string, string>(config);
 
  // 发送业务消息
  // 读取文件 读取内存数据库 读socket端口
  for (int i = 1; i <= 100; i++) {
   thread.sleep(500);
   producer.send(new keyedmessage<string, string>(config.topic,
     "this number ===>>> " + i));
  }
 
 }
}

三、编写消息者类consumerdemo

这个类的主要作用就是消费kafka中wordcount话题的消息。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
package com.lya.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;
 
/**
 * 消费者实例
 * @author liuyazhuang
 *
 */
public class consumerdemo {
  
 
 public static void main(string[] args) {
   
  properties props = new properties();
  props.put("zookeeper.connect", "192.168.209.121:2181");
  props.put("group.id", "1111");
  props.put("auto.offset.reset", "smallest");
  props.put("zk.connectiontimeout.ms", "15000");
 
  consumerconfig config = new consumerconfig(props);
  consumerconnector consumer =consumer.createjavaconsumerconnector(config);
  map<string, integer> topiccountmap = new hashmap<string, integer>();
  topiccountmap.put(config.topic, config.threads);
  map<string, list<kafkastream<byte[], byte[]>>> consumermap = consumer.createmessagestreams(topiccountmap);
  list<kafkastream<byte[], byte[]>> streams = consumermap.get(config.topic);
   
  for(final kafkastream<byte[], byte[]> kafkastream : streams){
   new thread(new runnable() {
    @override
    public void run() {
     for(messageandmetadata<byte[], byte[]> mm : kafkastream){
      string msg = new string(mm.message());
      system.out.println(msg);
     }
    }
    
   }).start();
   
  }
 }
}

四、运行实例

首先,运行消费者类consumerdemo
运行结果如下:

Kafka简单客户端编程实例

没有打印任何信息。
此时,我们运行生产者类producerdemo
我们再次打开消费者的控制台查看如下:

Kafka简单客户端编程实例

打印出了生产者生产的消息。
至此,kafka简单客户端编程实例结束。

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持服务器之家。

原文链接:http://blog.csdn.net/l1028386804/article/details/78383783