kafka消费者配置centos怎么调
在CentOS上配置Kafka消费者涉及多个步骤,包括安装必要的软件、配置Kafka和Zookeeper、创建和订阅主题,以及编写和运行消费者代码。以下是一个详细的指南:
1. 安装Java环境
首先,确保你的系统上安装了Java运行时环境。你可以使用以下命令来安装OpenJDK 8:
sudo yum install -y java-1.8.0-openjdk
java -version
2. 安装Zookeeper
下载并解压Zookeeper:
wget https://archive.apache.org/dist/zookeeper/zookeeper-3.4.6/zookeeper-3.4.6.tar.gz
tar xvf zookeeper-3.4.6.tar.gz
sudo mv zookeeper-3.4.6 /usr/local/zookeeper
sudo cat > /usr/local/zookeeper/conf/zoo.cfg << EOF
tickTime=2000
dataDir=/var/lib/zookeeper
clientPort=2181
EOF
sudo /usr/local/zookeeper/bin/zkServer.sh start
验证Zookeeper是否启动成功:
sudo netstat -nap | grep 2181
3. 安装Kafka
下载并解压Kafka:
wget https://downloads.apache.org/kafka/3.2.0/kafka_2.13-3.2.0.tgz
tar -xzvf kafka_2.13-3.2.0.tgz
sudo mv kafka_2.13-3.2.0 /usr/local/kafka
sudo mkdir /tmp/kafka-logs
配置Kafka:
sudo nano /usr/local/kafka/config/server.properties
确保以下配置正确:
listeners=PLAINTEXT://your_server_ip:9092
zookeeper.connect=localhost:2181
启动Kafka服务:
sudo /usr/local/kafka/bin/zookeeper-server-start.sh ../config/zookeeper.properties
sudo /usr/local/kafka/bin/kafka-server-start.sh ../config/server.properties
验证Kafka是否启动成功:
sudo netstat -an | grep 9092
4. 创建和订阅主题
使用Kafka命令行工具创建一个主题并订阅它。
创建主题:
sudo /usr/local/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
列出主题:
sudo /usr/local/kafka/bin/kafka-topics.sh --list --zookeeper localhost:2181
5. 编写和运行消费者代码
你可以使用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) {
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("auto.offset.reset", "earliest");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
KafkaConsumer consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("test"));
try {
while (true) {
ConsumerRecords 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();
}
}
}
编译并运行消费者程序:
javac -cp /usr/local/kafka/libs/*: SimpleConsumer.java
java -cp /usr/local/kafka/libs/*:. SimpleConsumer
请注意,上述步骤假设你已经有了一个运行的Kafka集群。如果你是在本地机器上运行单节点Kafka,上述步骤应该足够了。
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若内容造成侵权请联系我们,一经查实立即删除!