如何在面向请求的应用程序中管理Kafka事务性生产者对象

Question

当配置为事务生产者时，在面向请求（例如http或RPC服务器）的应用程序中管理Kafka生产者对象的最佳实践是什么？具体来说，如何在服务线程之间共享生产者对象，以及如何定义事务。这些对象的id配置值？

在非事务性使用中，生产者对象是线程安全的，并且在所有请求服务线程之间共享一个对象是很常见的。设置kafka消费者线程使用的事务性生产者对象也很简单，只需为每个消费者线程实例化一个对象即可工作得很好。

将事务生产者与面向请求的应用程序相结合似乎更为复杂，因为服务线程的生命周期通常由线程池动态控制。我可以想出几个选择，但都有缺点：

1. 共享单个对象，通过某种互斥来防止并发。负载下的争用可能是一个严重的问题
2. 为每个传入的请求实例化生产者对象。KafkaProducer对象初始化缓慢，因为它们维护网络连接、线程和其他重量级对象；为每个请求支付此费用似乎不切实际
3. 维护生产者对象池，并为每个请求租用一个。我能看到的主要缺点是所需的机器数量。还不清楚如何配置事务。对于这些对象，因为它们的生命周期没有像文档中所说的那样清晰地映射到分区、有状态的应用程序中的碎片标识符

还有其他选择吗？有最优方法吗？

Answer 1

事务性id用于防止由读取-处理-写入模式中的僵尸进程引起的重复，您从该模式中读取并生成kafka主题。对于面向请求的应用程序，例如由传入的超文本传输协议请求生成的消息，事务性id不会带来任何好处（当然，如果您想使用事务，您仍然需要分配一个，并且不应该在相同进程或集群中不同进程的生产者之间重复）

正如文档所说，事务生产者不是线程安全的

正如示例中所暗示的，每个生产者只能有一个未结交易。在beginTransaction（）和commitTransaction（）调用之间发送的所有消息都将是单个事务的一部分

因此，正如您正确解释的那样，不能并发访问生产者，因此我们必须从您描述的三个选项中选择一个。

对于这个答案，我将假设面向请求的应用程序对应于超文本传输协议请求，因为该机制正在触发与事务一起生成的消息（实际上，不止一条消息，否则对于幂等生产者来说就足够了，不需要事务）

就正确性而言，所有这些都是可以的，因为选项1可以工作，但根据您的应用程序吞吐量，它可能有很高的争用，选项2也可以工作，但您将付出更高延迟的代价，并且效率不高。恕我直言，我认为选项3可能是最好的，因为它是前两个选项之间的折衷方案，尽管当然需要更仔细的实施，而不是每次都打开一个新的生产者。

剩下的问题是如何将事务id分配给生产者，特别是在最后一种情况下（尽管选项1和3都有相同的问题，因为在这两种情况下，我们都在重用具有相同事务id的生产者来处理不同的请求）。

要回答这个问题，我们首先需要了解事务处理的目标。id是为了保护我们免受僵尸进程（一个挂起一段时间的进程，例如长gc暂停的bc，被视为已死亡，但一段时间后返回并继续）导致的重复消息的影响，这被称为僵尸Geofence。

理解僵尸Geofence需求的一个重要细节是理解它可能发生在哪个用例中，这是一种读取-处理-写入模式，在该模式中，您从主题中读取、处理元素并写入输出主题和偏移主题，这为我们提供了原子性和精确一次语义（如果您没有对流程步骤产生任何副作用）。幂等生产者防止了由于生产者重试（其中消息由代理持久化，但生产者未接收到ack）和kafka内的两阶段提交（其中我们不仅写入输出，而且还将消息标记为通过也生成到偏移主题来消费）而导致的重复防止了由于消费消息而导致的重复多次（如果进程在生成输出主题后但在提交偏移量之前崩溃）。还有一种微妙的情况是，可以引入副本，它是一个僵尸生产者，它通过每次生产者调用initTransactions时单调增加一个历元来进行隔离，该事务将与生产者发送的每条消息一起发送。因此，对于要隔离的制作人，另一个制作人应该使用相同的事务id启动，这里的关键由Jason Gustafson在本次演讲中解释

“我们正在寻找的是保证每个输入分区只有一个写入负责读取该数据并写入输出”

这意味着交易。id是根据“读-处理-写”模式中正在使用的分区分配的。因此，如果分配了主题a的分区0的流程被认为是死的，则重新平衡将开始，分配的新流程应创建具有相同事务的生产者。id，这就是为什么它应该是这样的

回到您最初的问题，http请求不会遵循僵尸可以引入重复项的读取过程写入模式，因为每个http请求都是唯一的，即使您引入了唯一标识符，从事务生产者的角度来看，它也将是不同的消息。在这种情况下，我认为，如果您希望向两个不同主题写入数据的原子性，那么使用事务生产者仍然有价值，但您可以为选项2选择一个随机事务id，或者将其重新用于选项1和3。

我的答案已经过时了，因为它基于Kafka的旧版本。KIP-447解决了前面描述的每个分区有一个生产者的开销问题

随着输入分区数量的增加，这种架构的扩展性不好。每个生产者都有单独的内存缓冲区、单独的线程和单独的网络连接。这限制了生产者的性能，因为我们无法有效地使用多个任务的输出来改进批处理。由于存在更多并发事务和更多冗余元数据管理，这也会给代理带来不必要的负载。

这是本文中解释的主要区别

在使用者组重新平衡后完成分区分配时，使用者的第一步是始终获取下一个偏移量以开始获取数据。通过这一观察，OffsetFetch协议的保护得到了增强，这样当使用者组具有与一个分区相关联的挂起事务偏移时，OffsetFetch调用可以被阻止，直到相关事务完成。以前，将返回“过时”的偏移数据，并允许应用程序立即继续。

有了这个新功能，我不再清楚transactional.id的使用。

尽管目前还不清楚为什么如果有待处理的交易，Geofence需要阻止投票，而在我看来，发送消费者组元数据就足够了（我假设僵尸生产者将通过promise旧generation.id来Geofencegroup.id，generation.id随着每次重新平衡而受到冲击），但transactional.id似乎不再扮演重要角色

在模式V1中，如果另一个实例具有相同的事务性，则生产者被“隔离”。id已启动。Spring通过为每个组使用一个制作人来管理这一点。id/主题/分区；当发生重新平衡时，新实例将使用相同的事务。id和老制片人被隔离<对于模式V2，每个组不需要一个生产者。id/topic/partition，因为消费者元数据与偏移量一起发送到事务，代理可以使用该信息确定生产者是否被隔离。