Kafka使用Java客户端进行访问

添加maven依赖包

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>0.9.0.1</version>
</dependency>
<dependency>
    <groupId>log4j</groupId>
    <artifactId>log4j</artifactId>
    <version>1.2.17</version>
</dependency>
<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-api</artifactId>
    <version>1.7.25</version>
</dependency>
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka_2.10</artifactId>
    <version>0.8.2.0</version>
</dependency>

建立包结构

　　建立包结构如下图所示为例：

　　在log4j.properties中输入：

log4j.rootLogger=INFO, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.Target=System.out
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{2}: %m%n

生产者代码

 1 package com.juyun.kafka;
 2
 3 import java.util.Properties;
 4
 5 import org.apache.log4j.PropertyConfigurator;
 6
 7 import kafka.javaapi.producer.Producer;
 8 import kafka.producer.KeyedMessage;
 9 import kafka.producer.ProducerConfig;
10 import kafka.serializer.StringEncoder;
11
12 public class KafkaProducerExample extends Thread {
13     private String topic;
14
15     public KafkaProducerExample(String topic){
16         super();
17         this.topic=topic;
18     }
19
20     @Override
21     public void run() {
22         Producer<Integer, String> producer=CreateProducer();
23         for (int i = 1; i < 10; i++) {
24             String message="message"+i;
25             producer.send(new KeyedMessage<Integer, String>(topic, message)); // 调用producer的send方法发送数据
26             System.out.println("发送："+message);
27             try {
28                 sleep(1000);
29             } catch (InterruptedException e) {
30                 e.printStackTrace();
31             }
32         }
33     }
34
35     public Producer<Integer, String> CreateProducer(){
36         Properties props=new Properties();
37         props.setProperty("zookeeper.connect", "172.16.0.157:2181"); // 与zookeeper建立连接
38         props.setProperty("serializer.class", StringEncoder.class.getName()); // key.serializer.class默认为serializer.class
39         props.setProperty("metadata.broker.list", "172.16.0.157:9092"); // kafka broker对应的主机，格式为host1:port1,host2:port2
40         props.put("request.required.acks","1"); // 等待topic中某个partition leader保存成功的状态反馈
41         Producer<Integer, String> producer = new Producer<Integer, String>(new ProducerConfig(props)); // 通过配置文件，创建生产者
42         return producer;
43     }
44
45     public static void main(String[] args){
46         PropertyConfigurator.configure("C:/Users/juyun/workspace/Kafka/src/main/java/com/juyun/logs/log4j.properties"); // 加载.properties文件
47         new KafkaProducerExample("test").start(); // 输入topic,启动线程
48     }
49
50 }  

KafkaProducerExample.java

消费者代码

 1 package com.juyun.kafka;
 2
 3 import java.util.HashMap;
 4 import java.util.List;
 5 import java.util.Map;
 6 import java.util.Properties;
 7
 8 import org.apache.log4j.PropertyConfigurator;
 9
10 import kafka.consumer.Consumer;
11 import kafka.consumer.ConsumerConfig;
12 import kafka.consumer.ConsumerIterator;
13 import kafka.consumer.KafkaStream;
14 import kafka.javaapi.consumer.ConsumerConnector;
15
16
17 public class KafkaConsumerExample extends Thread{
18     private String topic;
19
20     private KafkaConsumerExample(String topic) {
21         super();
22         this.topic=topic;
23     }
24
25     @Override
26     public void run() {
27         ConsumerConnector consumer = createConsumer(); // 创建消费者连接
28         Map<String,Integer> topicCountMap=new HashMap<String, Integer>(); // 定义一个map
29         topicCountMap.put(topic, 1);
30         // Map<String, List<KafkaStream<byte[], byte[]>> 中String是topic， List<KafkaStream<byte[], byte[]>是对应的流
31         Map<String, List<KafkaStream<byte[], byte[]>>> MessageStreams = consumer.createMessageStreams(topicCountMap);
32         // 取出 topic1对应的 streams
33         KafkaStream<byte[], byte[]> kafkaStream = MessageStreams.get(topic).get(0);
34         // 迭代获取到的流
35         ConsumerIterator<byte[], byte[]> iterator = kafkaStream.iterator();
36         while (iterator.hasNext()) {
37             String message = new String(iterator.next().message());
38             System.out.println("接收到："+message);
39         }
40     }
41
42     public ConsumerConnector createConsumer(){
43         Properties properties = new Properties();
44         properties.setProperty("zookeeper.connect", "172.16.0.157:2181");
45         properties.put("zookeeper.connectiontimeout.ms", "6000");
46         properties.setProperty("group.id", "group1"); // 设置这个消费者所在的group
47         // 只要ConsumerConnector还在的话，consumer会一直等待新消息，不会自己退出
48         ConsumerConnector createJavaConsumerConnector = Consumer.createJavaConsumerConnector(new ConsumerConfig(properties));
49         return createJavaConsumerConnector;
50     }
51
52     public static void main(String[] args) {
53         PropertyConfigurator.configure("C:/Users/juyun/workspace/Kafka/src/main/java/com/juyun/logs/log4j.properties"); // 加载.properties文件
54         new KafkaConsumerExample("test").start();
55     }
56 }  

KafkaConsumerExample.java

执行程序

　　需要先启动zookeeper

#进入到Zookeeper的bin目录下
cd /opt/zookeeper-3.4.8/bin
#启动服务
./zkServer.sh start

　　再启动Kafka

#进入到Kafka安装目录
bin/kafka-server-start.sh config/server.properties

　　并可以同时在命令终端启动生产者和消费者进行检测

#启动生产者
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic topic

#启动消费者
bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic topic --from-beginning