centos kafka如何配置生产者和消费者

在CentOS上配置Kafka的生产者和消费者，需要确保已经安装了Java和Kafka。以下是配置步骤：

生产者配置

1. 编辑生产者配置文件： 打开Kafka安装目录下的config/producer.properties文件。

2. 设置必要的参数：

   • bootstrap.servers: Kafka集群的地址列表，例如localhost:9092。
   • key.serializer: 键的序列化器，通常是org.apache.kafka.common.serialization.StringSerializer。
   • value.serializer: 值的序列化器，通常是org.apache.kafka.common.serialization.StringSerializer。

   示例配置：

   bootstrap.servers=localhost:9092
   key.serializer=org.apache.kafka.common.serialization.StringSerializer
   value.serializer=org.apache.kafka.common.serialization.StringSerializer
   

3. 保存并关闭文件。

消费者配置

1. 编辑消费者配置文件： 打开Kafka安装目录下的config/consumer.properties文件。

2. 设置必要的参数：

   • bootstrap.servers: Kafka集群的地址列表，例如localhost:9092。
   • group.id: 消费者组的ID，用于标识一组共享相同消费偏移量的消费者。
   • key.deserializer: 键的反序列化器，通常是org.apache.kafka.common.serialization.StringDeserializer。
   • value.deserializer: 值的反序列化器，通常是org.apache.kafka.common.serialization.StringDeserializer。

   示例配置：

   bootstrap.servers=localhost:9092
   group.id=my-consumer-group
   key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
   value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
   

3. 保存并关闭文件。

使用Kafka命令行工具测试

生产者测试

使用Kafka自带的命令行生产者发送消息：

kafka-console-producer.sh --broker-list localhost:9092 --topic my-topic

在提示符下输入消息并按回车键发送。

消费者测试

使用Kafka自带的命令行消费者接收消息：

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --from-beginning

这将从头开始消费主题my-topic的所有消息。

使用Java代码测试

如果你希望通过Java代码测试生产者和消费者，可以参考以下示例：

生产者示例

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class KafkaProducerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        KafkaProducer producer = new KafkaProducer<>(props);

        ProducerRecord record = new ProducerRecord("my-topic", "Hello, Kafka!");

        producer.send(record);
        producer.close();
    }
}

消费者示例

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.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;

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

public class KafkaConsumerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

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

        while (true) {
            ConsumerRecords 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());
            });
        }
    }
}

编译并运行这些Java程序，你应该能够看到生产者和消费者之间的消息传递。

希望这些步骤能帮助你在CentOS上成功配置和使用Kafka的生产者和消费者。

以上就是关于“centos kafka如何配置生产者和消费者”的相关介绍，筋斗云是国内较早的云主机应用的服务商，拥有10余年行业经验，提供丰富的云服务器、租用服务器等相关产品服务。云服务器资源弹性伸缩，主机vCPU、内存性能强悍、超高I/O速度、故障秒级恢复；电子化备案，提交快速，专业团队7×24小时服务支持！

简单好用、高性价比云服务器租用链接：https://www.jindouyun.cn/product/cvm