最全Java对接kafka简单示例_java 连接kafka，又到一年金三银四

最后

我想问下大家当初选择做程序员的初衷是什么？有思考过这个问题吗？高薪？热爱？

既然入了这行就应该知道，这个行业是靠本事吃饭的，你想要拿高薪没有问题，请好好磨练自己的技术，不要抱怨。有的人通过培训可以让自己成长，有些人可以通过自律强大的自学能力成长，如果你两者都不占，还怎么拿高薪？

架构师是很多程序员的职业目标，一个好的架构师是不愁所谓的35岁高龄门槛的，到了那个时候，照样大把的企业挖他。为什么很多人想进阿里巴巴，无非不是福利待遇好以及优质的人脉资源，这对个人职业发展是有非常大帮助的。

如果你也想成为一名好的架构师，那或许这份Java核心架构笔记你需要阅读阅读，希望能够对你的职业发展有所帮助。

中高级开发必知必会：

本文已被CODING开源项目：【一线大厂Java面试题解析+核心总结学习笔记+最新讲解视频+实战项目源码】收录

需要这份系统化的资料的朋友，可以点击这里获取

import java.util.Properties;

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






**目录**


[Java对接Kafka简单示例](#Java%E5%AF%B9%E6%8E%A5Kafka%E7%AE%80%E5%8D%95%E7%A4%BA%E4%BE%8B)


[步骤一：引入Kafka客户端库](#%E6%AD%A5%E9%AA%A4%E4%B8%80%EF%BC%9A%E5%BC%95%E5%85%A5Kafka%E5%AE%A2%E6%88%B7%E7%AB%AF%E5%BA%93)


[步骤二：创建Kafka生产者](#%E6%AD%A5%E9%AA%A4%E4%BA%8C%EF%BC%9A%E5%88%9B%E5%BB%BAKafka%E7%94%9F%E4%BA%A7%E8%80%85)


[步骤三：创建Kafka消费者](#%E6%AD%A5%E9%AA%A4%E4%B8%89%EF%BC%9A%E5%88%9B%E5%BB%BAKafka%E6%B6%88%E8%B4%B9%E8%80%85)


[总结](#%E6%80%BB%E7%BB%93)




---


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​`​中添加以下依赖项：



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

xmlCopy code
org.apache.kafka
kafka-clients
2.8.0

### 步骤二：创建Kafka生产者


接下来，我们需要创建一个Kafka生产者来发送消息。首先，我们需要设置Kafka的相关配置，如Kafka服务器的地址和端口号，以及消息的序列化方式等。然后，我们可以创建一个​`​Producer​`​对象，并使用​`​send()​`​方法来发送消息。以下是一个简单的示例代码：



1
2
3
4
5
6
7
8

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()​`​方法来获取消息。以下是一个简单的示例代码：



1
2
3
4
5
6
7
8

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的使用，我们可以更好地应用它来处理实时数据流和构建大规模的分布式系统。





## 最后

针对最近很多人都在面试，我这边也整理了相当多的面试专题资料，也有其他大厂的面经。希望可以帮助到大家。

> 下面的面试题答案都整理成文档笔记。也还整理了一些面试资料&最新2021收集的一些大厂的面试真题（都整理成文档，小部分截图）
>

![在这里插入图片描述](https://img-blog.csdnimg.cn/img_convert/4045fff78f6d3df39797d3281be22d5e.webp?x-oss-process=image/format,png)

> 最新整理电子书

![在这里插入图片描述](https://img-blog.csdnimg.cn/img_convert/5f23a84dca1b150ee7571a3a58620258.webp?x-oss-process=image/format,png)

> **本文已被[CODING开源项目：【一线大厂Java面试题解析+核心总结学习笔记+最新讲解视频+实战项目源码】](https://bbs.csdn.net/forums/4f45ff00ff254613a03fab5e56a57acb)收录**

**[需要这份系统化的资料的朋友，可以点击这里获取](https://bbs.csdn.net/forums/4f45ff00ff254613a03fab5e56a57acb)**

试题解析+核心总结学习笔记+最新讲解视频+实战项目源码】](https://bbs.csdn.net/forums/4f45ff00ff254613a03fab5e56a57acb)收录**

**[需要这份系统化的资料的朋友，可以点击这里获取](https://bbs.csdn.net/forums/4f45ff00ff254613a03fab5e56a57acb)**

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31