详解kafka消息发送重试机制的案例

news2024/11/16 17:33:56

在 Kafka 生产者中实现消息发送的重试机制，可以通过配置 KafkaProducer 的相关属性来实现。以下是一些关键的配置项：

retries：设置生产者发送失败后重试的次数。

retry.backoff.ms：设置生产者在重试前等待的时间。

buffer.memory：设置生产者在内存中缓存数据的最大值，如果达到这个值，生产者会拒绝接受新的消息，直到当前缓存的消息被发送出去。

batch.size：设置生产者在发送批次中可以包含的最大消息数。

linger.ms：设置生产者在发送批次之前等待更多消息的最大时间。

max.in.flight.requests.per.connection：设置每个连接最多数未完成的请求

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class KafkaProducerDemo {
    public static void main(String[] args) {
        // 配置生产者属性
        Properties props = new Properties();
        props.put("bootstrap.servers", "4.5.8.4:9092");
        props.put("key.serializer", StringSerializer.class.getName());
        props.put("value.serializer", StringSerializer.class.getName());
        props.put("retries", 5); // 设置重试次数
        props.put("retry.backoff.ms", 100); // 设置重试间隔
        props.put("buffer.memory", 33554432); // 设置缓冲区大小
        props.put("batch.size", 16384); // 设置批次大小
        props.put("linger.ms", 1); // 设置等待时间
        props.put("max.in.flight.requests.per.connection", 5); // 设置最大在途请求数

        // 创建生产者实例
        Producer<String, String> producer = new KafkaProducer<>(props);

        // 发送消息
        for (int i = 0; i < 1000000; i++) {
            String key = "案例1=====" + i;
            System.out.println("key:"+key);
            String value = "Spring AI Alibaba 实现了与阿里云通义模型的完整适配，接下来，我们将学习如何使用 spring ai alibaba 开发一个基于通义模型服务的智能聊天应用" + i;
            ProducerRecord<String, String> record = new ProducerRecord<>("test-topic", key, value);
            producer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    // 处理消息发送失败的情况
                    System.err.println("发送消息失败：" + exception.getMessage());
                } else {
                    // 处理消息发送成功的情况
                    System.out.println("消息发送成功，偏移量：" + metadata.offset());
                }
            });
        }

        // 关闭生产者
        producer.close();
    }
}

在这个示例中，我们设置了重试次数、重试间隔、缓冲区大小、批次大小、等待时间和最大在途请求数。此外，我们还为 send 方法提供了一个回调函数，用于处理消息发送成功或失败的情况。这样，当消息发送失败时，生产者会自动重试，直到达到配置的重试次数。如果所有重试都失败，回调函数会收到异常通知，你可以在回调中实现进一步的错误处理逻辑。

🔍 如何配置Kafka生产者的重试策略？

其实上面也有说，再次总结下

要配置 Kafka 生产者的重试策略，你可以按照以下步骤进行：

设置重试次数：
- 通过设置 retries 属性来指定生产者在遇到错误时重试发送消息的次数。例如，设置 retries 为 3 表示生产者会尝试最多 3 次发送消息。
设置重试间隔：
- 使用 retry.backoff.ms 属性来配置重试之间的时间间隔。这个设置可以防止生产者在连续的短时间内发送大量重试请求，给 Kafka 集群或网络造成压力。
确保消息幂等性：
- 设置 enable.idempotence 为 true 以确保生产者发送消息的逻辑是幂等的，即使消息被重复发送也不会影响系统状态。
配置确认策略：
- 通过 acks 属性来确保消息被所有副本确认。例如，设置 acks 为 “all” 可以确保消息被所有副本确认后才认为是成功发送。
异步发送与回调：
- 使用异步发送消息，并在回调中处理发送失败的情况。在回调中对异常进行分类处理，对于可恢复的错误进行重试，对于不可恢复的错误进行日志记录或报警。
错误处理与日志记录：
- 在回调函数中捕获并处理异常，同时记录详细的错误日志，便于问题排查和监控。
监控与告警：
- 对生产者的关键性能指标进行监控，如发送延迟、吞吐量等。当指标出现异常时，及时触发告警通知相关人员处理。
合理配置重试机制：
- 根据业务需求合理配置重试次数和重试间隔，以减少因网络波动或 Kafka 集群短暂不可用导致的消息丢失风险。
设置最大在途请求：
- 通过 max.in.flight.requests.per.connection 属性限制每个连接最多数未完成的请求，这有助于控制内存使用和重试的并发量。
配置超时时间：
- Kafka 2.4 版本引入了 delivery.timeout.ms 参数，它设置了发送记录和接收确认之间的超时时间。这个参数与 retries 结合使用，可以提供更灵活的重试控制。