如何在Ubuntu上配置Kafka消费者 - 问答

在Ubuntu上配置Kafka消费者涉及几个步骤，包括安装Java（如果尚未安装）、下载并解压Kafka、启动Zookeeper和Kafka服务器，以及编写和运行Kafka消费者。以下是详细步骤：

1. 安装Java

Kafka需要Java环境，确保你的系统上安装了Java。你可以使用以下命令安装OpenJDK：

sudo apt update
sudo apt install openjdk-11-jdk

验证安装：

java -version

2. 下载并解压Kafka

从Apache Kafka官方网站下载最新版本的Kafka，并解压到你的本地目录。

wget https://downloads.apache.org/kafka/3.4.0/kafka_2.13-3.4.0.tgz
tar -xzf kafka_2.13-3.4.0.tgz
cd kafka_2.13-3.4.0

3. 启动Zookeeper

Kafka使用Zookeeper进行集群管理。首先启动Zookeeper服务器：

bin/zookeeper-server-start.sh config/zookeeper.properties

4. 启动Kafka服务器

在另一个终端窗口中启动Kafka服务器：

bin/kafka-server-start.sh config/server.properties

5. 创建一个主题

创建一个Kafka主题，以便消费者可以订阅：

bin/kafka-topics.sh --create --topic test-topic --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1

6. 编写Kafka消费者代码

你可以使用Java编写一个简单的Kafka消费者。以下是一个示例代码：

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class SimpleConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "localhost:9092");
        props.setProperty("group.id", "test-group");
        props.setProperty("key.deserializer", StringDeserializer.class.getName());
        props.setProperty("value.deserializer", StringDeserializer.class.getName());
        props.setProperty("auto.offset.reset", "earliest");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("test-topic"));

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                records.forEach(record -> {
                    System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
                });
            }
        } finally {
            consumer.close();
        }
    }
}

7. 编译并运行消费者代码

将上述代码保存为SimpleConsumer.java，然后编译并运行：

javac -cp $(find . -name "*.jar") SimpleConsumer.java
java -cp .:$(find . -name "*.jar") SimpleConsumer

注意：在Windows上，你需要使用分号（;）而不是冒号（:）来分隔类路径。

8. 验证消费者

确保Kafka服务器和Zookeeper正在运行，并且主题已经创建。你应该能够看到消费者从主题中读取消息并打印出来。

通过以上步骤，你可以在Ubuntu上成功配置并运行一个Kafka消费者。

0 赞

0 踩