java生产消费kafka

在Java中，可以使用 Kafka 客户端来实现生产者和消费者，将数据发送到 Kafka 和从 Kafka 接收数据。以下是一个简单的示例代码，演示如何在Java中实现 Kafka 生产者和消费者：

Kafka 生产者：

import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; import java.util.Properties;

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

    Properties props = new Properties();
    props.put("bootstrap.servers", bootstrapServers);
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

    KafkaProducer<String, String> producer = new KafkaProducer<>(props);

    try {
        for (int i = 0; i < 10; i++) {
            String message = "Message " + i;
            ProducerRecord<String, String> record = new ProducerRecord<>(topicName, message);
            producer.send(record);
            System.out.println("Sent: " + message);
            Thread.sleep(1000);
        }
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        producer.close();
    }
}

}

Kafka 消费者：

import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.KafkaConsumer; import java.util.Arrays; import java.util.Properties;

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

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

    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
    consumer.subscribe(Arrays.asList(topicName));

    try {
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("Received: " + record.value());
            }
        }
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        consumer.close();
    }
}

}

在上面的示例中，我们使用 KafkaProducer 类实现了一个简单的 Kafka 生产者，并使用 KafkaConsumer 类实现了一个简单的 Kafka 消费者。请注意，具体的配置参数和主题名称需要根据你的 Kafka 配置进行调整。

注意：本文归作者所有，未经作者允许，不得转载