Java Kafka实现消息的生产和消费

需求
在项目开发中需要往Kafka中存放图片数据，另外一个程序需要从Kafka中获取图片数据，进行图片分析。

引入依赖

<dependency>
    <groupId>org.apache.kafkagroupId>
    <artifactId>kafka-clientsartifactId>
    <version>2.1.0version>
dependency>
1
2
3
4
5

消息生产者
bootstrap.servers：kafka服务地址和端口，例如172.16.xxx.xxx:9092
topic：主题用于消息分类，例如test2

import java.util.Properties;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

public static void sendMsg() {
    Properties properties = new Properties();
    properties.put("bootstrap.servers", "172.16.xxx.xxx:9092");
    properties.put("acks", "all");
    properties.put("retries", 0);
    properties.put("batch.size", 16384);
    properties.put("linger.ms", 1);
    properties.put("buffer.memory", 33554432);
    properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    Producer<String, String> producer = null;
    try {
        producer = new KafkaProducer<>(properties);
        String msg1 = "http://5b0988e595225.cdn.sohucs.com/images/20190314/d0fed36f3bfe4bf2b83c5f8e0e9e2252.jpg";
        producer.send(new ProducerRecord<>("test2", msg1));
        System.out.println("Sent:" + msg1);
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        producer.close();
    }
}
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

消息消费者
bootstrap.servers：kafka服务地址和端口，例如172.16.xxx.xxx:9092
topic：主题用于消息分类，例如test2

private static void consumerMsg() {
    Properties properties = new Properties();
    properties.put("bootstrap.servers", "172.16.xxx.xxx:9092");
    properties.put("group.id", "group-1");
    properties.put("enable.auto.commit", "true");
    properties.put("auto.commit.interval.ms", "1000");
    properties.put("auto.offset.reset", "earliest");
    properties.put("session.timeout.ms", "30000");
    properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

    KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
    kafkaConsumer.subscribe(Arrays.asList("test2"));
    while (true) {
        ConsumerRecords<String, String> records = kafkaConsumer.poll(10000);
        for (ConsumerRecord<String, String> record : records) {
            System.out.printf("offset = %d, value = %s", record.offset(), record.value());
            System.out.println();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21