1、异步发送消息&回调

回调就是接收kafka的响应

1.1、pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>3.0.5</version><relativePath/> <!-- lookup parent from repository --></parent><!-- Generated by https://start.springboot.io --><!-- 优质的 spring/boot/data/security/cloud 框架中文文档尽在 => https://springdoc.cn --><groupId>com.atguigu.kafka</groupId><artifactId>kafka-producer</artifactId><version>0.0.1-SNAPSHOT</version><name>kafka-producer</name><description>kafka-producer</description><properties><java.version>17</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>3.0.0</version></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>

java_60">1.2、KafkaProducer1.java

java">package com.atguigu.kafka.producer;
import org.apache.kafka.clients.producer.*;
import java.util.Properties;
public class KafkaProducer1 {/*** 主函数用于演示如何向Kafka的特定主题发送消息。** @param args 命令行参数（未使用）*/public static void main(String[] args) {// 初始化Kafka生产者配置Properties props = new Properties();props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.74.148:9092"); // 指定Kafka broker的地址和端口props.put("acks", "all"); // 确认消息写入策略props.put("retries", 0); // 消息发送失败时的重试次数props.put("linger.ms", 1); // 发送缓冲区等待时间props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 指定键的序列化器props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 指定值的序列化器// 创建Kafka生产者实例Producer<String, String> producer = new KafkaProducer<>(props);// 发送消息到主题"my_topic3"// 异步发送消息：不接收kafka的响应//producer.send(new ProducerRecord<String, String>("my_topic3",  "hello,1,2,3"));// 注释掉的循环代码块展示了如何批量发送消息//for (int i = 0; i < 100; i++)//    producer.send(new ProducerRecord<String, String>("my-topic", Integer.toString(i), Integer.toString(i)));producer.send(new ProducerRecord<String, String>("my_topic3",  "hello_3"),new Callback() {//消息发送成功，kafka broker ack 以后回调@Overridepublic void onCompletion(RecordMetadata recordMetadata, Exception e) {//exception：如果有异常代表消息未能正常发送到kafka，没有异常代表消息发送成功：//此时kafka的消息不一定持久化成功（需要kafka生产者加配置）//RecordMetadata代表发送成功的消息的元数据System.out.println("recordMetadata.topic() = " + recordMetadata.topic());System.out.println("recordMetadata.partition() = " + recordMetadata.partition());System.out.println("msg offset = " + recordMetadata.offset());System.out.println("recordMetadata.timestamp() = " + recordMetadata.timestamp());}});// 关闭生产者实例producer.close();}
}

recordMetadata.topic() = my_topic3
recordMetadata.partition() = 1
msg offset = 1
recordMetadata.timestamp() = 1716976982243