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

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

生产者配置

编辑生产者配置文件：打开Kafka安装目录下的config/producer.properties文件。
设置必要的参数：
- 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
```
保存并关闭文件。

消费者配置

编辑消费者配置文件：打开Kafka安装目录下的config/consumer.properties文件。
设置必要的参数：
- 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
```
保存并关闭文件。

使用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的生产者和消费者。