Java对接kafka简单示例_java 连接kafka

作者：小丑西瓜9 | 2024-04-09 16:58:25

踩

java 连接kafka

Java可以使用Apache Kafka提供的kafka-clients库来对接Kafka。下面是一个简单的示例代码，展示了如何使用Java对接Kafka并发送和接收消息：首先，确保已经在项目中添加了kafka-clients库的依赖。


import org.apache.kafka.clients.producer.*;
import org.apache.kafka.clients.consumer.*;
import java.util.Properties;
public class KafkaExample {
    private static final String TOPIC = "test-topic";
    private static final String BOOTSTRAP_SERVERS = "localhost:9092";
    
    public static void main(String[] args) {
        // 生产者示例
        Properties producerProps = new Properties();
        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        
        Producer<String, String> producer = new KafkaProducer<>(producerProps);
        
        String message = "Hello, Kafka!";
        ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, message);
        
        producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (exception != null) {
                    System.err.println("Failed to send message: " + exception.getMessage());
                } else {
                    System.out.printf("Sent message: topic=%s, partition=%d, offset=%d%n",
                            metadata.topic(), metadata.partition(), metadata.offset());
                }
            }
        });
        
        producer.close();
        
        // 消费者示例
        Properties consumerProps = new Properties();
        consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        Consumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
        consumer.subscribe(Collections.singletonList(TOPIC));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("Received message: topic=%s, partition=%d, offset=%d, key=%s, value=%s%n",
                        record.topic(), record.partition(), record.offset(), record.key(), record.value());
            }
        }
    }
}

以上代码演示了如何使用Kafka的生产者将消息发送到指定的topic，以及如何使用消费者从指定的topic接收消息。请根据实际情况修改TOPIC和BOOTSTRAP_SERVERS变量的值。

如果你使用Maven构建你的项目，可以在项目的pom.xml文件中添加以下依赖来引入Kafka客户端库：


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

然后，你可以在你的Java代码中使用Kafka的API来对接Kafka。以下是一个示例的pom.xml文件，展示了如何添加Kafka依赖：


<project>
    <!-- 其他配置 -->
    <dependencies>
        <!-- Kafka依赖 -->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>2.8.0</version>
        </dependency>
    </dependencies>
    <!-- 其他配置 -->
</project>

请注意，上面的示例使用了Kafka的2.8.0版本，你可以根据实际情况选择合适的版本。添加依赖后，你可以在你的Java代码中使用Kafka的API，如上面的示例代码所示。记得在你的Java文件中引入相关的类，例如：


import org.apache.kafka.clients.producer.*;
import org.apache.kafka.clients.consumer.*;
import java.util.Properties;

这样，你就可以使用Maven管理你的项目的依赖，并使用Kafka的API对接Kafka。

Kafka是一种分布式流处理平台，最初由LinkedIn开发并开源。它以高吞吐量、可扩展性和容错性为特点，用于处理大规模的实时数据流。 Kafka的设计目标是提供一种高效的、可持久化的、分布式发布-订阅消息系统。它采用了分布式、分区和复制的架构，可以同时处理大量的实时数据流，并将数据持久化存储在集群中，以便后续的数据分析和处理。 Kafka的核心概念包括以下几个部分：

Topic（主题）：消息的发布和订阅都是基于主题进行的，每个主题可以分为多个分区。
Partition（分区）：主题可以被划分为多个分区，每个分区是一个有序的日志文件，用于存储消息。分区可以分布在不同的Kafka集群节点上，实现数据的分布式存储和处理。
Producer（生产者）：生产者负责向Kafka集群发送消息，并将消息发布到指定的主题和分区。
Consumer（消费者）：消费者从Kafka集群订阅主题，并消费分区中的消息。消费者可以以不同的方式进行消息消费，如批量消费、实时消费等。
Broker（代理服务器）：Kafka集群中的每个节点都被称为代理服务器，负责存储和处理消息。
ZooKeeper（动物管理员）：Kafka使用ZooKeeper来进行集群的协调和管理，包括领导者选举、分区分配和集群状态的维护等。 Kafka广泛应用于大数据领域，特别适用于实时数据流处理、日志收集和数据管道等场景。它可以与其他大数据生态系统工具（如Hadoop、Spark、Flink等）无缝集成，实现高效的数据处理和分析。