避免重复Kafka生产者消息

Question

我正在使用Spring Boot中的Kafka模板。Java 8

我的主要目的是，消费者不应重复使用信息。

1）调用表获取100行并将其发送到kafka

2） 假设我处理了70行（我得到了成功确认），然后Kafka宕机了（Kafka在RETRY机制计时内无法恢复）

因此，当我重新启动Spring启动应用程序时，我如何确保不再发送这70条消息。

一种选择是我可以在数据库表消息 is_sent = Y 或 N 中使用标志。

还有其他有效的方法吗？

Answer 1

我将使用JDBC源连接器（取决于您当前使用的数据库）和Kafka Connect来正确处理此场景。

如果你仍然想编写自己的制作人，Kafka FAQ 的这一部分应该很有用：

我如何从Kafka那里得到一次信息？

恰好一次语义有两个部分:避免数据生产期间的重复和避免数据消费期间的重复。

有两种方法可以在数据生产过程中获得一次语义：

< li >对每个分区使用单写入程序，并且每次出现网络错误时，检查该分区中的最后一条消息，以查看您的最后一次写入是否成功 < li >在消息中包含一个主键(UUID或其他)，并在消费者端进行重复数据删除。

如果您执行其中一项操作，那么Kafka所承载的日志将不会重复。然而，没有重复的阅读也取决于消费者的一些合作。如果消费者定期检查其位置，那么如果它失败并重新启动，它将从检查点位置重新启动。因此，如果数据输出和检查点不是以原子方式写入的，那么这里也可能会出现重复。此问题是存储系统特有的。例如，如果您正在使用数据库，则可以在事务中一起提交这些数据库。LinkedIn编写的HDFS加载器Camus为Hadoop加载提供了类似的功能。另一种不需要事务的方法是使用主题/分区/偏移量组合来存储偏移量和加载的数据并进行重复数据消除。

我认为有两个改进会让这变得容易得多：

1. 通过可选地在服务器上集成对生产者幂等的支持，生产者幂等可以自动完成，而且成本更低
2. 现有的高级消费者没有暴露出很多更精细的偏移控制（例如，重置您的位置）。我们很快就会着手

Answer 2

对于Kafka，我看到了存储指向id的指针以跟踪您在主题中的位置的实现，以及使用某种分布式存储以在集群级别跟踪这一点的实现。我还没有做太多的工作，所以我将尝试提供一个我们与SQS一起用于重复检测的解决方案。很可能Kafka有一个更好的解决方案，这是一个解决重复问题的方案，只是想添加到那里，以便您也可以查看其他解决方案。

我在与AWS SQS合作点对点消息传递用例时遇到了同样的问题，因为它提供了至少一次交付保证，而不是一次且仅一次。

我们最终使用 Redis 及其分布式锁定策略来解决这个问题。我 https://angularthinking.blogspot.com/ 这里有一篇文章。

高级方法是创建一个分布式锁，将一个条目放入缓存中，并根据您的用例使用适当的TTL。我们使用LUA脚本执行putIfNotExists（）方法，如上面的博客所示。规模是我们关注的问题之一，通过上述实现，我们能够每秒处理10万条消息，而SQS和redis的规模都非常好。我们必须根据吞吐量和缓存增长将TTL调整为最佳值。我们确实有24小时或更短的复制窗口的好处，因此，根据redis进行此决定是可以的。如果您有更长的窗口，重复可能会在几天或几个月内发生，redis选项可能不适用。

我们还研究了DynamoDB来实现putIfNotExists()，但redis似乎在这个用例中性能更好，尤其是它使用LUA脚本实现了本机putIfNotExists。

祝你的搜索好运。