Kafka 数据重复怎么办?(案例)

这篇具有很好参考价值的文章主要介绍了Kafka 数据重复怎么办?(案例)。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

一、前言

数据重复这个问题其实也是挺正常,全链路都有可能会导致数据重复。

Kafka 数据重复怎么办?(案例)

通常,消息消费时候都会设置一定重试次数来避免网络波动造成的影响,同时带来副作用是可能出现消息重复。

整理下消息重复的几个场景:

  1. 生产端: 遇到异常,基本解决措施都是 重试
    • 场景一:leader分区不可用了,抛 LeaderNotAvailableException 异常,等待选出新 leader 分区。
    • 场景二:Controller 所在 Broker 挂了,抛 NotControllerException 异常,等待 Controller 重新选举。
    • 场景三:网络异常、断网、网络分区、丢包等,抛 NetworkException 异常,等待网络恢复。
  2. 消费端: poll 一批数据,处理完毕还没提交 offset ,机子宕机重启了,又会 poll 上批数据,再度消费就造成了消息重复。

怎么解决?

先来了解下消息的三种投递语义:

  • 最多一次(at most once): 消息只发一次,消息可能会丢失,但绝不会被重复发送。例如: mqttQoS = 0
  • 至少一次(at least once): 消息至少发一次,消息不会丢失,但有可能被重复发送。例如: mqttQoS = 1
  • 精确一次(exactly once): 消息精确发一次,消息不会丢失,也不会被重复发送。例如: mqttQoS = 2

了解了这三种语义,再来看如何解决消息重复,即如何实现精准一次,可分为三种方法:

  1. Kafka 幂等性 Producer 保证生产端发送消息幂等。局限性,是只能保证单分区且单会话(重启后就算新会话)
  2. Kafka 事务: 保证生产端发送消息幂等。解决幂等 Producer 的局限性。
  3. **消费端幂等: ** 保证消费端接收消息幂等。蔸底方案。

1)Kafka 幂等性 Producer

**幂等性指:**无论执行多少次同样的运算,结果都是相同的。即一条命令,任意多次执行所产生的影响均与一次执行的影响相同。

幂等性使用示例:在生产端添加对应配置即可

Properties props = new Properties();
props.put("enable.idempotence", ture);  // 1. 设置幂等
props.put("acks", "all");               // 2. 当 enable.idempotence 为 true,这里默认为 all
props.put("max.in.flight.requests.per.connection", 5); // 3. 注意
复制代码
  1. 设置幂等,启动幂等。
  2. 配置 acks,注意:一定要设置 acks=all,否则会抛异常。
  3. 配置 max.in.flight.requests.per.connection 需要 <= 5,否则会抛异常 OutOfOrderSequenceException
    • 0.11 >= Kafka < 1.1, max.in.flight.request.per.connection = 1
    • Kafka >= 1.1, max.in.flight.request.per.connection <= 5

为了更好理解,需要了解下 Kafka 幂等机制:

Kafka 数据重复怎么办?(案例)

  1. Producer 每次启动后,会向 Broker 申请一个全局唯一的 pid。(重启后 pid 会变化,这也是弊端之一)
  2. Sequence Numbe:针对每个 <Topic, Partition> 都对应一个从0开始单调递增的 Sequence,同时 Broker端会缓存这个 seq num
  3. 判断是否重复:<pid, seq num>Broker 里对应的队列 ProducerStateEntry.Queue(默认队列长度为 5)查询是否存在
    • 如果 nextSeq == lastSeq + 1,即 服务端seq + 1 == 生产传入seq,则接收。
    • 如果 nextSeq == 0 && lastSeq == Int.MaxValue,即刚初始化,也接收。
    • 反之,要么重复,要么丢消息,均拒绝。

Kafka 数据重复怎么办?(案例)

这种设计针对解决了两个问题:

  1. 消息重复: 场景 Broker 保存消息后还没发送 ack 就宕机了,这时候 Producer 就会重试,这就造成消息重复。
  2. 消息乱序: 避免场景,前一条消息发送失败而其后一条发送成功,前一条消息重试后成功,造成的消息乱序。

那什么时候该使用幂等:

  1. 如果已经使用 acks=all,使用幂等也可以。
  2. 如果已经使用 acks=0 或者 acks=1,说明你的系统追求高性能,对数据一致性要求不高。不要使用幂等。

2)Kafka 事务

使用 Kafka 事务解决幂等的弊端:单会话且单分区幂等。

Tips 这块篇幅较长,这先稍微提及下使用,之后另起一篇。

事务使用示例:分为生产端 和 消费端

Properties props = new Properties();
props.put("enable.idempotence", ture);  // 1. 设置幂等
props.put("acks", "all");               // 2. 当 enable.idempotence 为 true,这里默认为 all
props.put("max.in.flight.requests.per.connection", 5); // 3. 最大等待数
props.put("transactional.id", "my-transactional-id");  // 4. 设定事务 id

Producer<String, String> producer = new KafkaProducer<String, String>(props);

// 初始化事务
producer.initTransactions();

try{
    // 开始事务
    producer.beginTransaction();

    // 发送数据
    producer.send(new ProducerRecord<String, String>("Topic", "Key", "Value"));
 
    // 数据发送及 Offset 发送均成功的情况下,提交事务
    producer.commitTransaction();
} catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
    // 数据发送或者 Offset 发送出现异常时,终止事务
    producer.abortTransaction();
} finally {
    // 关闭 Producer 和 Consumer
    producer.close();
    consumer.close();
}
复制代码

这里消费端 Consumer 需要设置下配置:isolation.level 参数

  • read_uncommitted 这是默认值,表明 Consumer 能够读取到 Kafka 写入的任何消息,不论事务型 Producer 提交事务还是终止事务,其写入的消息都可以读取。如果你用了事务型 Producer,那么对应的 Consumer 就不要使用这个值。

  • read_committed 表明 Consumer 只会读取事务型 Producer 成功提交事务写入的消息。当然了,它也能看到非事务型 Producer 写入的所有消息。

3)消费端幂等

“如何解决消息重复?” 这个问题,其实换一种说法:就是如何解决消费端幂等性问题。

只要消费端具备了幂等性,那么重复消费消息的问题也就解决了。

典型的方案是使用:消息表,来去重:

Kafka 数据重复怎么办?(案例)

  • 上述栗子中,消费端拉取到一条消息后,开启事务,将消息Id 新增到本地消息表中,同时更新订单信息。
  • 如果消息重复,则新增操作 insert 会异常,同时触发事务回滚。


 

二、案例:Kafka 幂等性 Producer 使用

环境搭建可参考:链接

准备工作如下:

  1. Zookeeper:本地使用 Docker 启动

    $ docker run -d --name zookeeper -p 2181:2181 zookeeper
    a86dff3689b68f6af7eb3da5a21c2dba06e9623f3c961154a8bbbe3e9991dea4
    复制代码
  2. Kafka:版本 2.7.1,源码编译启动(看上文源码搭建启动)

  3. 启动生产者:Kafka 源码中 exmaple

  4. 启动消息者:可以用 Kafka 提供的脚本

    # 举个栗子:topic 需要自己去修改
    $ cd ./kafka-2.7.1-src/bin
    $ ./kafka-console-producer.sh --broker-list localhost:9092 --topic test_topic
    复制代码

创建 topic 1副本,2 分区

$ ./kafka-topics.sh --bootstrap-server localhost:9092 --topic myTopic --create --replication-factor 1 --partitions 2

# 查看
$ ./kafka-topics.sh --bootstrap-server broker:9092 --topic myTopic --describe
复制代码

生产者代码:

Kafka 数据重复怎么办?(案例)

public class KafkaProducerApplication {

    private final Producer<String, String> producer;
    final String outTopic;

    public KafkaProducerApplication(final Producer<String, String> producer,
                                    final String topic) {
        this.producer = producer;
        outTopic = topic;
    }

    public void produce(final String message) {
        final String[] parts = message.split("-");
        final String key, value;
        if (parts.length > 1) {
            key = parts[0];
            value = parts[1];
        } else {
            key = null;
            value = parts[0];
        }
        final ProducerRecord<String, String> producerRecord 
            = new ProducerRecord<>(outTopic, key, value);
        producer.send(producerRecord,
                (recordMetadata, e) -> {
                    if(e != null) {
                        e.printStackTrace();
                    } else {
                        System.out.println("key/value " + key + "/" + value + "\twritten to topic[partition] " + recordMetadata.topic() + "[" + recordMetadata.partition() + "] at offset " + recordMetadata.offset());
                    }
                }
        );
    }

    public void shutdown() {
        producer.close();
    }

    public static void main(String[] args) {

        final Properties props = new Properties();

        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
        props.put(ProducerConfig.ACKS_CONFIG, "all");

        props.put(ProducerConfig.CLIENT_ID_CONFIG, "myApp");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        final String topic = "myTopic";
        final Producer<String, String> producer = new KafkaProducer<>(props);
        final KafkaProducerApplication producerApp = new KafkaProducerApplication(producer, topic);

        String filePath = "/home/donald/Documents/Code/Source/kafka-2.7.1-src/examples/src/main/java/kafka/examples/input.txt";
        try {
            List<String> linesToProduce = Files.readAllLines(Paths.get(filePath));
            linesToProduce.stream().filter(l -> !l.trim().isEmpty())
                    .forEach(producerApp::produce);
            System.out.println("Offsets and timestamps committed in batch from " + filePath);
        } catch (IOException e) {
            System.err.printf("Error reading file %s due to %s %n", filePath, e);
        } finally {
            producerApp.shutdown();
        }
    }
}
复制代码

启动生产者后,控制台输出如下:

Kafka 数据重复怎么办?(案例)

启动消费者:

$ ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic myTopic
复制代码

Kafka 数据重复怎么办?(案例)

修改配置 acks

启用幂等的情况下,调整 acks 配置,生产者启动后结果是怎样的:

  • 修改配置 acks = 1
  • 修改配置 acks = 0

会直接报错:

Exception in thread "main" org.apache.kafka.common.config.ConfigException: Must set acks to all in order to use the idempotent producer. 
Otherwise we cannot guarantee idempotence.
复制代码

Kafka 数据重复怎么办?(案例)

修改配置 max.in.flight.requests.per.connection

启用幂等的情况下,调整此配置,结果是怎样的:

  • max.in.flight.requests.per.connection > 5 会怎样?

Kafka 数据重复怎么办?(案例)

当然会报错:

Caused by: org.apache.kafka.common.config.ConfigException: Must set max.in.flight.requests.per.connection to at most 5 to use the idempotent producer.
复制代码

Kafka 数据重复怎么办?(案例)文章来源地址https://www.toymoban.com/news/detail-417250.html

到了这里,关于Kafka 数据重复怎么办?(案例)的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 电脑提示数据错误循环冗余检查怎么办?

    有些时候,我们尝试在磁盘上创建分区或清理硬盘时,还可能会遇到这个问题:数据错误循环冗余检查。这是如何导致的呢?我们又该如何解决这个问题呢?下面我们就来了解一下。 数据错误循环冗余检查怎么解决呢?在回答这个问题之前我们需要先了解一下导致问题的原因

    2024年02月12日
    浏览(57)
  • 领导临时要数据库文档怎么办?

    很多时候,我们为了着急忙慌赶项目进度,很容易忽略整理文档这件事 某一天,领导心血来潮,要搞一次突击检查, 想看看我们的数据库设计的是否规范, 但他又不想亲自去数据库查验(毕竟这么大领导) 那么,我们该怎么办? 第一种方法:离职,世界那么大,我想去看

    2024年02月08日
    浏览(40)
  • 数据库同步 Elasticsearch 后数据不一致,怎么办?

    Q1:Logstash 同步 postgreSQL 到 Elasticsearch 数据不一致。 在使用 Logstash 从 pg 库中将一张表导入到 ES 中时,发现 ES 中的数据量和 PG 库中的这张表的数据量存在较大差距。如何快速比对哪些数据没有插入?导入过程中,Logstash 日志没有异常。PG 中这张表有 7600W。 Q2:mq 异步双写数

    2024年02月15日
    浏览(41)
  • mysql数据库忘记密码了怎么办

    本人用的mysql8版本 看到网上很多教程,什么修改配置文件my.ini。在8版本根本没用。以下是8版本解决办法。亲测可用。 1、用管理员身份打开命令行工具。(强调:管理员身份) 2、停止mysql服务: 3、输入以下命令无密码启动mysql 4、 另开一个命令行窗口,输入mysql -u root无密

    2024年02月11日
    浏览(46)
  • 数据分析过程中,发现数值缺失,怎么办?

    按照数据缺失机制,数据分析过程中,我们可以将其分为以下几类: (1)完全随机缺失(MCAR):所缺失的数据发生的概率既与已观察到的数据无关,也与未观察到的数据无关。 (2)随机缺失(MAR):假设缺失数据发生的概率与所观察到的变量是有关的,而与未观察到的数

    2024年02月05日
    浏览(52)
  • 服务器数据被盗了该怎么办

    随着互联网的发展,如今人们习惯了在互联网上分享生活、购物等等。便捷了人们的生活,也让互联网企业蓬勃生机,但同时也暗藏着危机。其中服务器被入侵是常见的黑客攻击方式,不仅会给企业带来经济损失,同时也让企业在公众面前失去了信誉。下面我们来看看一些案

    2024年02月04日
    浏览(61)
  • app渗透测试抓不到数据包怎么办?

    1、app渗透测试常见的模拟器有夜神,闪电模拟器等。 2、遇到问题,数据包抓不到? (1)最常见的解决方法调低模拟器安卓版本,使用MuMu模拟器,大部分app都适用,但是也有些app无法抓取。 (2)安装证书到系统根证书,可以解决上面的问题(一般情况下手机在安装了BurpSuite的伪证

    2024年02月13日
    浏览(38)
  • MySQL数据库忘记密码怎么办?教你一招

    文章目录 1.以管理员身份打开cmd,关闭Mysql服务 2. 跳过密码授权登录  3.再继续以管理员身份打开一个cmd窗口,进行重置密码  4.使用新密码重新登录mysql验证  5.使用Navicat可视化工具连接Mysql Mysql数据库之前安装好了,但是突然忘记当初自己设置的登录密码了,导致使用Navi

    2024年02月04日
    浏览(111)
  • CAD数据导入到ArcGIS中出现乱码怎么办?

    在项目中,我们通常会涉及到将甲方提供的CAD数据提取成标准.SHP数据,但是CAD它导入到ArcGIS Desktop中竟然乱码了,标注看不清了,好头大。其实是字符集的问题,将注册表中的字符集改为简体中文就可以解决,下面来看解决方法。 在windows搜索框中输入“regedit”。 点击“注册

    2024年02月05日
    浏览(47)
  • SQL Server 数据库变成单个用户怎么办

    参考技术A 1、首先我们打开SQL  SERVER的管理控制台,找到一个要设置角色的用户。 2、下面我们将为这个用户赋予创建数据库的角色,我们先用这个用户登录管理工具看一下是否具有创建用户的权限。 3、进行数据库创建的时候,提示如下的错误,证明这个用户不具备这个角色

    2024年02月03日
    浏览(56)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包