Kafka消费者在Linux上怎么配置 - 问答

在Linux上配置Kafka消费者涉及多个步骤，包括安装Java环境、下载和配置Kafka、编写消费者代码以及运行消费者。以下是一个详细的步骤指南：

1. 安装Java环境

Kafka是用Java编写的，因此需要在Linux系统上安装Java运行时环境（JRE）或Java开发工具包（JDK）。

sudo apt update
sudo apt install openjdk-11-jdk

验证安装：

java -version

2. 下载和配置Kafka

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

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

3. 启动Zookeeper和Kafka服务器

Kafka依赖于Zookeeper来管理集群状态。首先启动Zookeeper，然后启动Kafka服务器。

启动Zookeeper

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

启动Kafka服务器

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

4. 创建一个主题

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

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

5. 编写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 java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class SimpleConsumer {
    public static void main(String[] args) {
        String bootstrapServers = "localhost:9092";
        String groupId = "test-group";
        String topic = "test-topic";

        Properties props = new Properties();
        props.put("bootstrap.servers", bootstrapServers);
        props.put("group.id", groupId);
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

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

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

6. 编译和运行消费者程序

将上述代码保存为SimpleConsumer.java，然后编译和运行它。

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

确保find . -name "*.jar"命令返回所有Kafka客户端库的路径。

7. 验证消费者是否正常工作

在另一个终端窗口中，向主题发送一些消息以验证消费者是否正常工作。

bin/kafka-console-producer.sh --topic test-topic --bootstrap-server localhost:9092

输入一些消息并按回车键发送。

回到消费者程序的终端窗口，你应该能看到消费者接收到的消息。

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

0 赞

0 踩