卡夫卡连接节流答案

【问题标题】：Kafka connect throttling卡夫卡连接节流
【发布时间】：2017-03-02 23:43:57
【问题描述】：

我需要代表一组只公开 REST API 的懒惰消费者来使用消息。因此，我计划让 Sink 连接器从 Kafka 主题中获取消息，并对暴露的 API 执行 HTTP POST 操作。

要考虑的关键因素之一是节流。您建议使用什么机制来限制接收器任务以满足 API 的层 SLA。我知道 Kafka 具有客户端配额功能，但是，跟踪 API 请求/分钟或秒的最佳机制是什么，可以动态调整客户端配额？

【问题讨论】：

【解决方案1】：

我认为为您的 REST API 实施速率限制的最佳方法是在您的连接器代码中，如果需要，在 SinkTask.put() 中进行阻塞。您可能想考虑在您的SinkTasks 级别进行速率限制是否足够，或者您需要它是全局性的（由于涉及协调而更加复杂）。

您正在考虑使用 Kafka 配额的优点是可以为您处理分布式方面，但我相信目前只能根据传输的字节数进行配置。

【讨论】：

没错。 Kafka 客户端配额配置为传输的字节数，而 REST API 限制为否。的请求。就我而言，我将在云中的多个工作人员中运行多个接收器任务。当您说全局时，您是在暗示保持不计数。缓存（redis）之类的请求？有没有一种优雅的方式通过 Kafka connect 来实现这一点？
是的，使用 Redis 或 Memcached 之类的东西来维护限速 API 请求所需的共享状态。如果您可以接受可能过于保守的速率限制，您可以改为为您在put() 方法中获取的每个SinkTask 初始化一个番石榴RateLimiter。您授予RateLimiter 的许可数量可能是(overall_limit / num_tasks)（您可以在从SinkConnector.taskConfigs() 生成任务配置时传播任务级别限制）。
当使用 Guava RateLimiter 时，我相信我们需要在跨多个机架的多个工作人员运行的接收器任务中使用相同的 RateLimiter 实例。也就是说，任务 1 和任务 2 应该能够访问将调用 acquire() 方法的 RateLimiter 的单个实例。顺便说一句，我想你是在建议 Guava RateLimiter 作为缓存机制的替代品？
是的，使用每个任务 RateLimiter 可以替代在 Memcached/Redis/etc 中使用全局共享状态。您不应该假设任务在同一个工作人员上运行，因此我建议为每个任务实例保留一个RateLimiter，并根据任务数量在它们之间平均分配许可。如果工作均匀分布在任务中，这应该会很好。