kafka用java收发消息

Kafka 是一个分布式流处理平台，它允许你发布和订阅记录流，类似于消息队列或企业消息系统。下面我将为你展示如何使用 Java 来发送（生产者）和接收（消耗者）Kafka 中的消息。
1. 添加依赖

首先，你需要将 Kafka 客户端的依赖添加到你的 Java 项目中。如果你使用 Maven，可以添加以下依赖到你的 pom.xml 文件中：

1. <dependencies>
2.     <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
3.     <dependency>
4.         <groupId>org.apache.kafka</groupId>
5.         <artifactId>kafka-clients</artifactId>
6.         <version>你的Kafka客户端版本号</version>
7.     </dependency>
8.     <!-- 其他依赖... -->
9. </dependencies>

复制代码

确保将 你的Kafka客户端版本号 更换为当前你需要的 Kafka 客户端版本。
2. 发送消息（生产者）

以下是一个简单的 Kafka 生产者示例，用于向 Kafka 主题发送消息：

1. import org.apache.kafka.clients.producer.*;
3. import java.util.Properties;
5. public class KafkaProducerExample {
7.     public static void main(String[] args) {
8.         Properties props = new Properties();
9.         props.put("bootstrap.servers", "localhost:9092");
10.         props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
11.         props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
13.         Producer<String, String> producer = new KafkaProducer<>(props);
15.         for (int i = 0; i < 100; i++) {
16.             ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", Integer.toString(i), "Hello Kafka " + i);
17.             producer.send(record, new Callback() {
18.                 @Override
19.                 public void onCompletion(RecordMetadata metadata, Exception exception) {
20.                     if (exception != null) {
21.                         exception.printStackTrace();
22.                     } else {
23.                         System.out.printf("Record sent to partition %d with offset %d%n", metadata.partition(), metadata.offset());
24.                     }
25.                 }
26.             });
27.         }
29.         producer.close();
30.     }
31. }

复制代码

3. 接收消息（消耗者）

以下是一个简单的 Kafka 消耗者示例，用于从 Kafka 主题接收消息：

1. import org.apache.kafka.clients.consumer.*;
2. import org.apache.kafka.common.TopicPartition;
4. import java.time.Duration;
5. import java.util.Arrays;
6. import java.util.Collections;
7. import java.util.Properties;
9. public class KafkaConsumerExample {
11.     public static void main(String[] args) {
12.         Properties props = new Properties();
13.         props.put("bootstrap.servers", "localhost:9092");
14.         props.put("group.id", "test");
15.         props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
16.         props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
18.         Consumer<String, String> consumer = new KafkaConsumer<>(props);
19.         consumer.subscribe(Collections.singletonList("my-topic"));
21.         while (true) {
22.             ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
23.             for (ConsumerRecord<String, String> record : records) {
24.                 System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
25.             }
26.         }
27.     }
28. }

复制代码

注意：在实际应用中，你大概需要处理更多的异常和关闭资源，以及使用更复杂的配置和分区策略。上面的示例只是为了展示根本的使用方式。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。