kafka javaAPI入库程序的实现方法

1. 引入依赖

首先你需要在你的Java项目中引入Kafka的Java API的相关依赖。

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.8.0</version>
</dependency>

2. 创建Kafka生产者

接下来，你需要创建一个Kafka生产者实例，用于将数据写入Kafka集群。

// 引入依赖
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.KafkaProducer;
import java.util.Properties;

// 创建生产者实例
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");  // Kafka集群的地址
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");  // 键的序列化器
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");  // 值的序列化器
Producer<String, String> producer = new KafkaProducer<>(props);

// 发送消息
String topic = "my_topic";  // 发送的主题
String key = "my_key";  // 消息的键
String value = "Hello Kafka!";  // 消息的值
ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
producer.send(record);

// 关闭生产者实例
producer.close();

3. 创建Kafka消费者

如果你需要从Kafka中读取数据，你也需要创建一个Kafka消费者实例。

// 引入依赖
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Properties;
import java.util.Collections;

// 创建消费者实例
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");  // Kafka集群的地址
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");  // 键的反序列化器
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");  // 值的反序列化器
Consumer<String, String> consumer = new KafkaConsumer<>(props);

// 订阅主题
String topic = "my_topic";  // 订阅的主题
consumer.subscribe(Collections.singletonList(topic));

// 拉取消息
ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records) {
    String key = record.key();
    String value = record.value();
    System.out.println("Received message: " + value);
}

// 关闭消费者实例
consumer.close();

4. 完整程序示例

下面是一个完整的示例程序，展示了如何使用Kafka的Java API进行数据的生产和消费。

import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Properties;
import java.util.Collections;

public class KafkaExample {

    public static void main(String[] args) {
        // 生产者示例
        Properties producerProps = new Properties();
        producerProps.put("bootstrap.servers", "localhost:9092");
        producerProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        producerProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        Producer<String, String> producer = new KafkaProducer<>(producerProps);
        String topic = "my_topic";
        String key = "my_key";
        String value = "Hello Kafka!";
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
        producer.send(record);
        producer.close();

        // 消费者示例
        Properties consumerProps = new Properties();
        consumerProps.put("bootstrap.servers", "localhost:9092");
        consumerProps.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        consumerProps.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        Consumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
        String topic = "my_topic";
        consumer.subscribe(Collections.singletonList(topic));
        ConsumerRecords<String, String> records = consumer.poll(100);
        for (ConsumerRecord<String, String> record : records) {
            String key = record.key();
            String value = record.value();
            System.out.println("Received message: " + value);
        }
        consumer.close();
    }
}

本文由轻山版权所有，禁止未经同意的情况下转发