Kafka发送消息是异步发送的，所以我们不知道消息是否发送成功，所以会可能造成消息丢失。而且Kafka架构是由生产者-服务器端-消费者三种组成部分构成的。要保证消息不丢失，那么主要有三种解决方法。
生产者(producer)端处理
生产者默认发送消息代码如下：
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
public class KafkaMessageProducer {
public static void main(String[] args) {
// 配置Kafka生产者
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092"); // Kafka集群地址
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);
String topic = "test"; // Kafka主题
try {
// 发送消息到Kafka
for (int i = 0; i < 10; i++) {
String message = "Message " + i;
ProducerRecord<String, String> record = new ProducerRecord<>(topic, message);
producer.send(record);
System.out.println("Sent message: " + message);
}
} catch (Exception e) {
e.printStackTrace();
} finally {
// 关闭Kafka生产者
producer.close();
}
}
}生产者端要保证消息发送成功，可以有两个方法：
1.把异步发送改成同步发送，这样producer就能实时知道消息的发送结果。
要将 Kafka 发送方法改为同步发送，可以使用 `send()` 方法的返回值Future<RecordMetadata>`， 并调用 `get()` 方法来等待发送完成。
以下是将 Kafka 发送方法改为同步发送的示例代码：
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import org.apache.kafka.clients.producer.RecordMetadata;
public class KafkaMessageProducer {
public static void main(String[] args) {
// 配置 Kafka 生产者
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092"); // Kafka 集群地址
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);
String topic = "test"; // Kafka 主题
try {
// 发送消息到 Kafka
for (int i = 0; i < 10; i++) {
String message = "Message " + i;
ProducerRecord<String, String> record = new ProducerRecord<>(topic, message);
RecordMetadata metadata = producer.send(record).get(); // 同步发送并等待发送完成
System.out.println("Sent message: " + message + ", offset: " + metadata.offset());
}
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
} finally {
// 关闭 Kafka 生产者
producer.close();
}
}
}在这个示例代码中，通过调用 send(record).get() 实现了同步发送，其中 get() 方法会阻塞当前线程，直到发送完成并返回消息的元数据。
2.添加异步回调函数来监听消息发送的结果，如果发送失败，可以在回调函数里重新发送。
要保持发送消息成功并添加回调函数，你可以在发送消息的时候指定一个回调函数作为参数。回调 函数将在消息发送完成后被调用，以便你可以在回调函数中处理发送结果。
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.RecordMetadata;
public class KafkaMessageProducer {
public static void main(String[] args) {
// 配置 Kafka 生产者
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092"); // Kafka 集群地址
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);
String topic = "test"; // Kafka 主题
try {
// 发送消息到 Kafka
for (int i = 0; i < 10; i++) {
String message = "Message " + i;
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.out.println("Sent message: " + message + ", offset: " + metadata.offset());
} else {
// 这里重新发送消息
producer.send(record);
exception.printStackTrace();
}
}
});
}
} finally {
// 关闭 Kafka 生产者
producer.close();
}
}
}在这个示例代码中，我们使用了 send(record, callback) 方法来发送消息，并传递了一个实现了 Callback 接口的匿名内部类作为回调函数。当消息发送完成后，回调函数的 onCompletion() 方法会被调用。你可以根据 RecordMetadata 和 Exception 参数来处理发送结果。
另外producer还提供了一个重试参数，这个参数叫retries，如果因为网络问题或者Broker故障导致producer发送消息失败，那么producer会根据这个参数的值进行重试发送消息。
服务器端(Broker)端
Kafka Broker（服务器端）通过以下方式来确保生产者端消息发送的成功和不丢失：
1. 消息持久化（异步刷盘）：Kafka Broker将接收到的消息持久化到磁盘上的日志文件中。这样即使在消息发送后发生故障，Broker能够恢复并确保消息不会丢失。（注意：持久化是由操作系统调度的，如果持久化之前系统崩溃了，那么就因为不能持久化导致数据丢失，但是Kafka没提供同步刷盘策略）
2. 复制与高可用性：Kafka支持分布式部署，可以将消息分布到多个Broker上形成一个Broker集群。在集群中，消息被复制到多个副本中，以提供冗余和高可用性。生产者发送消息时，它可以将消息发送到任何一个Broker，然后Broker将确保消息在集群中的所有副本中都被复制成功。
3. 消息提交确认：当生产者发送消息后，在收到Broker的确认响应之前，生产者会等待。如果消息成功写入并复制到了指定的副本中，Broker会发送确认响应给生产者。如果生产者在指定的时间内没有收到确认响应，它将会尝试重新发送消息，以确保消息不会丢失。
4. 可靠性设置（同步刷盘）：生产者可以配置一些参数来提高消息发送的可靠性。例如，可以设置`acks`参数来指定需要收到多少个Broker的确认响应才认为消息发送成功。可以将`acks`设置为`"all"`，表示需要收到所有副本的确认响应才算发送成功。
总之，Kafka Broker通过持久化和复制机制，以及消息确认和可靠性设置，确保生产者端的消息发送成功且不丢失。同时，应注意及时处理可能的错误情况，并根据生产者端需求和场景合理配置相应的参数。
对于使用YAML文件进行Kafka配置的情况，你可以按照以下格式设置acks参数：
# Kafka生产者配置
producer:
bootstrap.servers: your-kafka-server:9092
acks: all # 设置acks参数为"all"
key.serializer: org.apache.kafka.common.serialization.StringSerializer
value.serializer: org.apache.kafka.common.serialization.StringSerializer消费者（Consumer）处理
Kafka Consumer 默认会确保消息的至少一次传递（at least once delivery）。这意味着当 Consumer 完成对一条消息的处理后，会向 Kafka 提交消息的偏移量（offset），告知 Kafka 这条消息已被成功处理。如果 Consumer 在处理消息时发生错误，可以通过回滚偏移量来重试处理之前的消息。
以下是一些确保消息消费成功的方法：
- 使用自动提交偏移量（Auto Commit Offsets）
- 手动提交偏移量（Manual Commit Offsets）
- 设置消费者的最大重试次数：
- 设置适当的消费者参数
尽管 Kafka 提供了可靠的消息传递机制，但仍然需要在消费者端实现适当的错误处理和重试逻辑，以处理可能发生的错误情况。
![[hddm]python模块hddm安装后测试代码](https://img-blog.csdnimg.cn/direct/94855141845b40529578290714fbeeb4.png)
