⚙️ 如何调整重试策略以适应不同的业务需求？

🕗 发布于 2024-11-13 15:32 数据库 kafka java

调整 Kafka 生产者和消费者的重试策略以适应不同的业务需求，需要根据业务的特性和容错要求来进行细致的配置。以下是一些关键的调整策略：

业务重要性：
- 对于关键业务消息，可以增加重试次数，并设置较长的重试间隔，以减少消息丢失的风险。
- 对于非关键业务消息，可以减少重试次数或不进行重试，以避免不必要的资源消耗。
消息幂等性：
- 如果业务逻辑是幂等的，即多次处理相同消息不会导致业务状态不一致，可以增加重试次数。
- 如果业务逻辑不是幂等的，需要谨慎设置重试策略，或者实现去重逻辑。
消息时效性：
- 对于时效性要求高的消息，可以减少重试间隔，以便快速尝试重新发送。
- 对于时效性要求不高的消息，可以增加重试间隔，减少对 Kafka 集群的压力。
系统容量和负载：
- 根据 Kafka 集群和下游系统的容量和负载情况调整重试策略，避免因重试导致的额外负载影响系统稳定性。
错误类型：
- 对于临时性错误（如网络问题），可以设置较高的重试次数和较短的重试间隔。
- 对于永久性错误（如消息格式错误），应减少重试次数，避免无意义的重试。
死信队列（DLQ）：
- 对于重试次数用尽后仍然发送失败的消息，可以配置死信队列进行存储，以便后续分析和处理。
监控和告警：
- 实施实时监控，对重试次数、失败率等关键指标进行监控，并设置告警阈值。
业务流程控制：
- 在业务流程中实现重试逻辑，例如在业务层捕获异常并根据业务规则进行重试。
自定义重试策略：
- 实现自定义的重试策略，例如指数退避策略，以适应特定的业务场景。
事务性消息：
- 如果业务要求消息发送的原子性，可以启用事务性消息发送，确保消息要么全部发送成功，要么全部不发送。
资源限制：
- 考虑到生产者和消费者的资源限制，如内存和网络带宽，合理设置重试策略，避免资源耗尽。
反馈机制：
- 建立反馈机制，根据业务运行情况和系统性能反馈调整重试策略。

通过综合考虑上述因素，可以为不同的业务需求定制合适的重试策略，以确保 Kafka 消息系统的高效性和可靠性。

在这里插入图片描述

以下是一些代码案例，展示了如何根据不同的业务需求调整 Kafka 生产者和消费者的重试策略

在这里插入图片描述

Kafka 生产者重试策略案例

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 CustomRetryProducerDemo {
    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", 1000); // 设置重试间隔为1秒
        props.put("buffer.memory", 33554432); // 设置缓冲区大小
        props.put("batch.size", 16384); // 设置批次大小
        props.put("linger.ms", 10); // 设置等待时间为10毫秒
        props.put("max.in.flight.requests.per.connection", 1); // 设置最大在途请求数

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

        // 发送消息
        for (int i = 0; i < 100; i++) {
            String key = "key-" + i;
            String value = "value-" + 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();
    }
}

Kafka 消费者重试策略案例

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class CustomRetryConsumerDemo {
    public static void main(String[] args) {
        // 配置消费者属性
        Properties props = new Properties();
        props.put("bootstrap.servers", "4.5.8.4:9092");
        props.put("group.id", "test-group");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        props.put("max.poll.records", 500); // 设置每次拉取的最大记录数
        props.put("fetch.min.bytes", 1024); // 设置最小获取1KB的数据
        props.put("fetch.max.wait.ms", 500); // 设置最大等待500ms

        // 创建消费者实例
        Consumer<String, String> consumer = new KafkaConsumer<>(props);

        // 订阅主题
        consumer.subscribe(Collections.singletonList("test-topic"));

        // 消费消息
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                try {
                    // 处理消息
                    System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
                    // 假设处理消息可能会失败
                    if (record.value().contains("error")) {
                        throw new RuntimeException("模拟处理消息失败");
                    }
                } catch (Exception e) {
                    // 处理消息失败，记录日志或重试
                    System.err.println("处理消息失败：" + e.getMessage());
                    // 可以在这里实现重试逻辑，例如将消息发送到死信队列
                }
            }
            // 批量提交偏移量
            consumer.commitSync();
        }
    }
}

死信队列（DLQ）案例

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 DLQProducerDemo {
    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", 1000); // 设置重试间隔为1秒

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

        // 发送消息
        for (int i = 0; i < 100; i++) {
            String key = "key-" + i;
            String value = "value-" + i;
            ProducerRecord<String, String> record = new ProducerRecord<>("test-topic", key, value);
            producer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    // 处理消息发送失败的情况
                    System.err.println("发送消息失败：" + exception.getMessage());
                    // 将失败的消息发送到死信队列
                    ProducerRecord<String, String> dlqRecord = new ProducerRecord<>("test-topic-DLQ", key, exception.getMessage());
                    producer.send(dlqRecord);
                } else {
                    // 处理消息发送成功的情况
                    System.out.println("消息发送成功，偏移量：" + metadata.offset());
                }
            });
        }

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

这些代码案例展示了如何根据不同的业务需求调整 Kafka
生产者和消费者的重试策略，包括设置重试次数、重试间隔、处理消息发送失败的情况以及实现死信队列（DLQ）。希望这些示例能帮助您更好地理解和应用
Kafka 的重试机制。

原文地址：https://blog.csdn.net/weixin_44372802/article/details/143720331

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：ML 系列：第 23 节 — 离散概率分布（多项式分布）
下一篇：Python | Leetcode Python题解之第557题反转字符串中的单词III

C#调试项目_附加进程调试
c#项目附加进程一级目录二级目录三级目录一级目录二级目录三级目录
阅读更多2024-11-15
线程-2-线程概念与控制
线程概念与控制，局部性原理， POXIS接口，分页式存储结构，页表结构，缺页中断
阅读更多2024-11-15
洛谷 P4011 孤岛营救问题（BFS分层图最短路，状态压缩）
我们考虑对已获得的钥匙进行二进制状态压缩。注意：同一个格子可能有多把钥匙，一开始的。显然，我们直接使用BFS求最短路即可。这个格子，且当前已有钥匙的状态为。
阅读更多2024-11-15
聊天服务器(3)muduo网络库
muduo只能装在linux中，依赖boost库客户端并不需要高并发。
阅读更多2024-11-15
Unity学习---IL2CPP打包时可能遇到的问题
在这种情况下一般我们把Managed Stripping Level设置为High，然后看一下报错的提示，往link.wxl文件中添加要保留的东西即可。IL2CPP打包后会自动对Unity工程的dll
阅读更多2024-11-15
OpenGL C++视频中添加图片及文字水印播放并录制
根据前置储备，该功能的实现流程：创建一个GLThread的OpenGL运行环境；YUV数据绘制到OpenGL纹理中；图片纹理创建绘制到glViewport；文字渲染绘制到glViewport；录制视频
阅读更多2024-11-15
java实现中小企业的erp系统
java实现中小企业的erp系统
阅读更多2024-11-15
智能零售柜商品识别
本项目在智能零售商品识别的背景下，采用了多项技术创新和优化，旨在提高商品检测与识别的效率和准确性。首先，使用了PaddleX作为训练框架，这是一个高效、灵活的深度学习平台，简化了模型训练和部署的复杂性
阅读更多2024-11-15
H.265流媒体播放器EasyPlayer.js网页直播/点播播放器WebGL: CONTEXT_LOST_WEBGL错误引发的原因
EasyPlayer.js播放器不仅支持H.264与H.265视频编码格式，还具备直播、点播、录像、快照截图、MP4播放、多屏播放、倍数播放、全屏播放等功能特性，兼容Windows、Linux、And
阅读更多2024-11-15
ubuntu连接orangepi-zero-2w桌面的几种方法
线连接 Orange Pi 开发板和 HDMI 显示器。然后打开 Nomachine 图形化连接开发板。例如使用finalshell 连接开发板。其中视频采集卡是usb输出，hdmi输入。线 +
阅读更多2024-11-15

⚙️ 如何调整重试策略以适应不同的业务需求？

以下是一些代码案例，展示了如何根据不同的业务需求调整 Kafka 生产者和消费者的重试策略

Kafka 生产者重试策略案例

Kafka 消费者重试策略案例

死信队列（DLQ）案例

相关文章