Kafka：Consumer訂閱

測試代碼

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

此時

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

輸出是符合預期的，因為這不是第一次釋出消息到這幾個

Topic

上，部落客之前測試過幾次。再來示範

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> topicPartitionList) {
        KafkaConsumer<String, String> consumer = createConsumer();
        consumer.assign(topicPartitionList);
        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);
    }
}

通過下面這行代碼即可訂閱多個

Partition

：

consumer.assign(topicPartitionList);

運作

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

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

Kafka：Consumer訂閱

測試代碼

繼續閱讀

nginx location中斜線的位置的重要性

Apache2.4.x 配置檔案詳解Apache配置需要了解如下：開始講解：

配置apache支援PHP（win7）

27 Best Free Eclipse Plug-ins for Java Developer to be ProductiveCode Quality PluginsText Editor PluginsDependency ManagementVersion Control Integration PluginsFramework Development Continuous Integration Related PluginsOther Utility Plugins

Java String.format方法的簡單使用

neo4j之cypher使用文檔

GitHub連夜封殺！這份阿裡 10W 字内部 Java 字面試手冊到底有多強？

spark/scala關于【資源檔案】加載方法概述外部檔案加載方案測試資源檔案打包入jar包中小結

mybatis_入門程式Mybatis入門

AOP程式設計_Android優雅權限架構(1)概念基礎，2021金三銀四前言正文大綱正文

Effective Java 8:通用程式設計

OOM三種類型

工廠模式-三種類型

【遞歸】高效率求2的n次幂

win10本地scala和spark安裝安裝scala安裝spark

scala (3) Function 和 Method