0
0

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的核心概念包括以下几个部分：

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

Java对接Kafka简单示例

Kafka是一种高吞吐量的分布式消息队列系统，常用于处理大规模的实时数据流。在Java中，我们可以使用Kafka提供的客户端库来对接Kafka，并进行消息的发送和接收。以下是一个简单的Java对接Kafka的示例。

步骤一：引入Kafka客户端库

首先，我们需要在Java项目中引入Kafka客户端库。可以通过Maven或Gradle等构建工具来添加Kafka依赖项。例如，使用Maven可以在​

​pom.xml​

​中添加以下依赖项：

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

步骤二：创建Kafka生产者

接下来，我们需要创建一个Kafka生产者来发送消息。首先，我们需要设置Kafka的相关配置，如Kafka服务器的地址和端口号，以及消息的序列化方式等。然后，我们可以创建一个​

​Producer​

​对象，并使用​

​send()​

​方法来发送消息。以下是一个简单的示例代码：

javaCopy codeimport org.apache.kafka.clients.producer.*;
import java.util.Properties;
public class KafkaProducerExample {
    public static void main(String[] args) {
        // 设置Kafka配置
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "localhost:9092");
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // 创建Kafka生产者
        Producer<String, String> producer = new KafkaProducer<>(properties);
        // 发送消息
        String topic = "example-topic";
        String key = "key1";
        String value = "Hello, Kafka!";
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
        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.println("Sent message: topic = " + metadata.topic() +
                            ", partition = " + metadata.partition() +
                            ", offset = " + metadata.offset());
                }
            }
        });
        // 关闭Kafka生产者
        producer.close();
    }
}

步骤三：创建Kafka消费者

除了发送消息，我们还可以创建一个Kafka消费者来接收消息。类似地，我们需要设置Kafka的相关配置，并使用​

​poll()​

​方法来获取消息。以下是一个简单的示例代码：

javaCopy codeimport org.apache.kafka.clients.consumer.*;
import java.util.Collections;
import java.util.Properties;
public class KafkaConsumerExample {
    public static void main(String[] args) {
        // 设置Kafka配置
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "localhost:9092");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("group.id", "example-group");
        // 创建Kafka消费者
        Consumer<String, String> consumer = new KafkaConsumer<>(properties);
        // 订阅主题
        String topic = "example-topic";
        consumer.subscribe(Collections.singletonList(topic));
        // 拉取消息
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("Received message: topic = " + record.topic() +
                        ", partition = " + record.partition() +
                        ", offset = " + record.offset() +
                        ", key = " + record.key() +
                        ", value = " + record.value());
            }
        }
    }
}

总结

通过以上示例，我们可以看到Java如何对接Kafka并进行简单的消息发送和接收。使用Kafka可以实现高吞吐量的消息处理，并且具有良好的可扩展性和容错性。通过学习Kafka的使用，我们可以更好地应用它来处理实时数据流和构建大规模的分布式系统。

标签： java kafka
本文转载自: https://blog.csdn.net/q7w8e9r4/article/details/132025512
版权归原作者 牛肉胡辣汤 所有， 如有侵权，请联系我们删除。
登录后发布评论

“Java对接kafka简单示例”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

机器人技术在安全领域的重要作用

安全学习记录——网络篇（一）

图腾柱PFC：HP1010为您的电动两轮车之旅提供绿色，高效，安全的动力

单元测试数据库回滚问题

这才是单元测试，之前我们都错了

【C语言基础】：字符函数和字符串函数

ChatGPT-Next-Web SSRF漏洞+XSS漏洞复现(CVE-2023-49785)

文章导航