0
点赞
收藏
分享

微信扫一扫

Kafka：Consumer订阅

眼君 2022-02-14 阅读 132
kafkazookeeper分布式

测试代码

pom.xml：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.kaven</groupId>
    <artifactId>kafka</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>3.0.0</version>
        </dependency>
    </dependencies>
</project>

创建Topic：

package com.kaven.kafka.admin;

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.KafkaFuture;

import java.util.Collections;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;

public class Admin {

    // 基于Kafka服务地址与请求超时时间来创建AdminClient实例
    private static final AdminClient adminClient = Admin.getAdminClient(
            "192.168.1.9:9092,192.168.1.9:9093,192.168.1.9:9094",
            "40000");

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        Admin admin = new Admin();
        // 创建Topic，Topic名称为topic1，分区副本数为1，复制因子为1
        admin.createTopic("topic1", 1, (short) 1);
        // 创建Topic，Topic名称为topic2，分区副本数为2，复制因子为1
        admin.createTopic("topic2", 2, (short) 1);
        // 创建Topic，Topic名称为topic3，分区副本数为2，复制因子为1
        admin.createTopic("topic3", 2, (short) 1);
        Thread.sleep(10000);
    }

    public static AdminClient getAdminClient(String address, String requestTimeoutMS) {
        Properties properties = new Properties();
        properties.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, address);
        properties.setProperty(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, requestTimeoutMS);
        return AdminClient.create(properties);
    }

    public void createTopic(String name, int numPartitions, short replicationFactor) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(1);
        CreateTopicsResult topics = adminClient.createTopics(
                Collections.singleton(new NewTopic(name, numPartitions, replicationFactor))
        );
        Map<String, KafkaFuture<Void>> values = topics.values();
        values.forEach((name__, future) -> {
            future.whenComplete((a, throwable) -> {
                if(throwable != null) {
                    System.out.println(throwable.getMessage());
                }
                System.out.println(name__);
                latch.countDown();
            });
        });
        latch.await();
    }
}

Producer发布消息：

package com.kaven.kafka.producer;

import org.apache.kafka.clients.producer.*;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class ProducerTest {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        send("topic1");
        send("topic2");
        send("topic3");
    }

    public static void send(String name) throws ExecutionException, InterruptedException {
        Producer<String, String> producer = ProducerTest.createProducer();
        for (int i = 0; i < 7; i++) {
            ProducerRecord<String, String> producerRecord = new ProducerRecord<>(
                    name,
                    "key-" + i,
                    "value-" + i
            );
            // 异步发送并回调
            producer.send(producerRecord, (metadata, exception) -> {
                if(exception == null) {
                    System.out.printf("topic: %s, partition: %s, offset: %s\n", name, metadata.partition(), metadata.offset());
                }
                else {
                    exception.printStackTrace();
                }
            });
        }
        // 要关闭Producer实例
        producer.close();
    }

    public static Producer<String, String> createProducer() {
        // Producer的配置
        Properties properties = new Properties();
        // 服务地址
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.9:9092,192.168.1.9:9093,192.168.1.9:9094");
        // KEY的序列化器类
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // VALUE的序列化器类
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        return new KafkaProducer<>(properties);
    }
}

Consumer可以订阅多个Topic或者多个Partition，先来演示订阅多个Topic：

package com.kaven.kafka.consumer;

import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.*;

public class ConsumerTest {

    public static void main(String[] args) {
        subscribeTopicList(Arrays.asList("topic1", "topic2", "topic3"));
    }

    public static void subscribeTopicList(List<String> topicList) {
        KafkaConsumer<String, String> consumer = createConsumer();
        consumer.subscribe(topicList);
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(10000));
            records.forEach((record) -> {
                System.out.printf("topic: %s, partition: %s, offset: %s, key: %s, value: %s\n",
                        record.topic(), record.partition(), record.offset(), record.key(), record.value());
            });
        }
    }

    public static KafkaConsumer<String, String> createConsumer() {
        // Consumer的配置
        Properties properties = new Properties();
        // 服务地址
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.9:9092,192.168.1.9:9093,192.168.1.9:9094");
        // 组ID，用于标识此消费者所属的消费者组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "kaven-test");
        // 开启offset自动提交
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        // 消费者offset自动提交到Kafka的频率（以毫秒为单位）
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        // KEY的反序列化器类
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        // VALUE的反序列化器类
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        return new KafkaConsumer<>(properties);
    }
}

通过下面这行代码即可订阅多个Topic：

consumer.subscribe(topicList);

先创建这几个Topic。
在这里插入图片描述
再启动Consumer订阅程序，现在还没有消息。
在这里插入图片描述

然后通过Producer发布消息到这几个Topic。
在这里插入图片描述

topic: topic1, partition: 0, offset: 21
topic: topic1, partition: 0, offset: 22
topic: topic1, partition: 0, offset: 23
topic: topic1, partition: 0, offset: 24
topic: topic1, partition: 0, offset: 25
topic: topic1, partition: 0, offset: 26
topic: topic1, partition: 0, offset: 27
topic: topic2, partition: 0, offset: 12
topic: topic2, partition: 0, offset: 13
topic: topic2, partition: 0, offset: 14
topic: topic2, partition: 0, offset: 15
topic: topic2, partition: 1, offset: 9
topic: topic2, partition: 1, offset: 10
topic: topic2, partition: 1, offset: 11
topic: topic3, partition: 0, offset: 12
topic: topic3, partition: 0, offset: 13
topic: topic3, partition: 0, offset: 14
topic: topic3, partition: 0, offset: 15
topic: topic3, partition: 1, offset: 9
topic: topic3, partition: 1, offset: 10
topic: topic3, partition: 1, offset: 11

输出是符合预期的，因为这不是第一次发布消息到这几个Topic上，博主之前测试过几次。此时Consumer就可以订阅到消息了，输出如下所示：

topic: topic1, partition: 0, offset: 21, key: key-0, value: value-0
topic: topic1, partition: 0, offset: 22, key: key-1, value: value-1
topic: topic1, partition: 0, offset: 23, key: key-2, value: value-2
topic: topic1, partition: 0, offset: 24, key: key-3, value: value-3
topic: topic1, partition: 0, offset: 25, key: key-4, value: value-4
topic: topic1, partition: 0, offset: 26, key: key-5, value: value-5
topic: topic1, partition: 0, offset: 27, key: key-6, value: value-6
topic: topic2, partition: 1, offset: 9, key: key-0, value: value-0
topic: topic2, partition: 1, offset: 10, key: key-3, value: value-3
topic: topic2, partition: 1, offset: 11, key: key-4, value: value-4
topic: topic2, partition: 0, offset: 12, key: key-1, value: value-1
topic: topic2, partition: 0, offset: 13, key: key-2, value: value-2
topic: topic2, partition: 0, offset: 14, key: key-5, value: value-5
topic: topic2, partition: 0, offset: 15, key: key-6, value: value-6
topic: topic3, partition: 0, offset: 12, key: key-1, value: value-1
topic: topic3, partition: 0, offset: 13, key: key-2, value: value-2
topic: topic3, partition: 0, offset: 14, key: key-5, value: value-5
topic: topic3, partition: 0, offset: 15, key: key-6, value: value-6
topic: topic3, partition: 1, offset: 9, key: key-0, value: value-0
topic: topic3, partition: 1, offset: 10, key: key-3, value: value-3
topic: topic3, partition: 1, offset: 11, key: key-4, value: value-4

再来演示Consumer订阅多个Partition：

package com.kaven.kafka.consumer;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.*;

public class ConsumerTest {

    public static void main(String[] args) {
        subscribeTopicPartitionList(Arrays.asList(
                new TopicPartition("topic1", 0),
                new TopicPartition("topic2", 0),
                new TopicPartition("topic3", 0),
                new TopicPartition("topic3", 1)
        ));
    }

    public static void subscribeTopicPartitionList(List<TopicPartition> topicList) {
        KafkaConsumer<String, String> consumer = createConsumer();
        consumer.assign(topicList);
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(10000));
            records.partitions().forEach((partition) -> {
                List<ConsumerRecord<String, String>> recordsWithPartition = records.records(partition);
                recordsWithPartition.forEach((record) -> {
                    System.out.printf("topic: %s, partition: %s, offset: %s, key: %s, value: %s\n",
                            record.topic(), record.partition(), record.offset(), record.key(), record.value());
                });
            });
        }
    }

    public static KafkaConsumer<String, String> createConsumer() {
        // Consumer的配置
        Properties properties = new Properties();
        // 服务地址
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.9:9092,192.168.1.9:9093,192.168.1.9:9094");
        // 组ID，用于标识此消费者所属的消费者组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "kaven-test");
        // 开启offset自动提交
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        // 消费者offset自动提交到Kafka的频率（以毫秒为单位）
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        // KEY的反序列化器类
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        // VALUE的反序列化器类
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        return new KafkaConsumer<>(properties);
    }
}

运行Consumer订阅程序，然后再通过Producer发布消息到这几个Topic。

topic: topic1, partition: 0, offset: 35
topic: topic1, partition: 0, offset: 36
topic: topic1, partition: 0, offset: 37
topic: topic1, partition: 0, offset: 38
topic: topic1, partition: 0, offset: 39
topic: topic1, partition: 0, offset: 40
topic: topic1, partition: 0, offset: 41
topic: topic2, partition: 1, offset: 15
topic: topic2, partition: 1, offset: 16
topic: topic2, partition: 1, offset: 17
topic: topic2, partition: 0, offset: 20
topic: topic2, partition: 0, offset: 21
topic: topic2, partition: 0, offset: 22
topic: topic2, partition: 0, offset: 23
topic: topic3, partition: 0, offset: 20
topic: topic3, partition: 0, offset: 21
topic: topic3, partition: 0, offset: 22
topic: topic3, partition: 0, offset: 23
topic: topic3, partition: 1, offset: 15
topic: topic3, partition: 1, offset: 16
topic: topic3, partition: 1, offset: 17

此时Consumer就可以订阅到消息了，输出如下所示：

topic: topic1, partition: 0, offset: 35, key: key-0, value: value-0
topic: topic1, partition: 0, offset: 36, key: key-1, value: value-1
topic: topic1, partition: 0, offset: 37, key: key-2, value: value-2
topic: topic1, partition: 0, offset: 38, key: key-3, value: value-3
topic: topic1, partition: 0, offset: 39, key: key-4, value: value-4
topic: topic1, partition: 0, offset: 40, key: key-5, value: value-5
topic: topic1, partition: 0, offset: 41, key: key-6, value: value-6
topic: topic2, partition: 0, offset: 20, key: key-1, value: value-1
topic: topic2, partition: 0, offset: 21, key: key-2, value: value-2
topic: topic2, partition: 0, offset: 22, key: key-5, value: value-5
topic: topic2, partition: 0, offset: 23, key: key-6, value: value-6
topic: topic3, partition: 0, offset: 20, key: key-1, value: value-1
topic: topic3, partition: 0, offset: 21, key: key-2, value: value-2
topic: topic3, partition: 0, offset: 22, key: key-5, value: value-5
topic: topic3, partition: 0, offset: 23, key: key-6, value: value-6
topic: topic3, partition: 1, offset: 15, key: key-0, value: value-0
topic: topic3, partition: 1, offset: 16, key: key-3, value: value-3
topic: topic3, partition: 1, offset: 17, key: key-4, value: value-4

输出也是符合预期的，Consumer订阅就介绍到这里，如果博主有说错的地方或者大家有不同的见解，欢迎大家评论补充。

举报

相关推荐

Kafka-Consumer

androidCameraView源码解析预览第三方库相机natario1

kafka——Consumer API

kafka公众号：阿区先生

Kafka核心之Consumer

kafka旧版新版本JavaScript前端开发

Kafka Consumer 位移提交

kafka

Kafka 源码分析——Consumer

react.js前端javascript

kafka Consumer pol模型

kafkadb

Kafka 的 Consumer Group 解读

kafka分布式

kafka-consumer-offset位移

kafka

python kafka consumer 内存控制

python内存管理消息处理Python后端开发

Kafka：Consumer手动提交offset

kafkazookeeperjava
0 条评论