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<
    String, String>
     producer = new KafkaProducer<
    >
    (props);
    

        ProducerRecord<
    String, String>
     record = new ProducerRecord<
    String, String>
    ("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<
    String, String>
     consumer = new KafkaConsumer<
    >
    (props);
    
        consumer.subscribe(Collections.singletonList("my-topic"));


        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());

            }
    );

        }

    }

}
    

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

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

声明：本文内容由网友自发贡献，本站不承担相应法律责任。对本内容有异议或投诉，请联系2913721942#qq.com核实处理，我们将尽快回复您，谢谢合作！