@codedreamstar 我大概讲一下场景出来吧。 应用 A 设计之初并没考虑到那么长远，初衷也是能消费多快就消费多快。因此就用上了多线程异步处理数据。 处理数据这块其实也只是为了把数据存到 redis 里。 然后我们有另外的进程去从 redis 的队列里拿到数据，然后把这些数据再下发到下游（通过调用下游接口，简称应用 B ）。 目前消费的 topic 都是推过来的实时数据，因此各项的 tps 都能够满足；不过应用 B 是有一个峰值的 tps 的。之前来了个需求，新接入一个 topic （简称(topic-new)，topic-new 推过来的量是固定的，我这边撑这块业务为：存量初始化。 之前协商好上游提供 topic 过来的时候是控制速率的（因此原本我这边不用考虑限速限流的），后来因沟通问题上游又不作限速处理，最终限速操作只能在应用 A 这边进行。
针对限速这块其实我是有过几个思考方案的
方案一：直接搞一个 Spark 应用来进行存量初始化，Spark 在控制批量消费还是很好控制的
方案二：使用令牌桶对应用 A 特殊的 Consumer 进行限流
方案三：对应用 A 的流入和流出都作限流操作（后续一定会排期对数据流出作限流操作，但是听各位大佬的建议，好像并不推荐对流入数据也作限流操作）
综合考虑各种因素，目前是考虑使用方案二进行限流操作，当完成存量初始化之后就可以下线该 topic 了，后续先实现流出的限流功能，其他功能再考虑可行性。

@codedreamstar 你好大佬，应用本身设计就是为了尽可能多消费来使用多线程实现的。 目前多线程主要是用来处理数据，且消费者处理的消息是无关的，提到 offset 提交，其实在 poll 到数据后，就先手动把 Offset 保存到 redis 里，然后配置 auto.commit.interval.ms=1 秒去自动提供 offset ，拿到的数据是直接丢到多线程里去异步处理了，应用不需要关注到当前批次的 records 处理完后才更新 offset 。这一点并不是很关注，主要是后续应用处理数据的时候会有各种机制把数据丢到 redis 里，成功的失败的处理都丢到 redis 里。

@ymz 感谢大佬提供 springboot 注解的思路，目前应用并不是依赖 springboot 框架搭建的，但后续是有升级到 springboot 框架的需求的。后续在应用需要迁移重构的时候，我会着重构思注解的可行性可实现方式。

@Takamine 应用里并不需要严格关注 kafka 自动提交 offset 与处理完 records 的数目一致。 目前设置的 auto.commit.interval.ms 是 1 秒，而且应用也有手动每秒往 redis 里写入当前读写的 offset 。

@flmn 你好，你提到的小坑，设置 max.poll.record 的同时是需要配合 fetch.min.bytes 使用是吧，我理解的是，如果一条数据本身不小，fetch.min.bytes 应该是有一个默认值，如果 max.poll.record*单条数据的大小 > fetch.min.bytes 默认值，实际还是按照默认值可获取的数量来获取吧。

@1Q1 目前我就是用谷歌 guava 的 RateLimiter 来限制指定时间最多 Poll 几次

@liuhan907 是的，我现在就是用令牌桶来实现 1 秒只 poll 一次，设定 poll 的最大数量来实现

@dd31san 感觉思路是可行的。不过现阶段 consumer 配置是 auto.commit 自动提交偏移量的。如果改成手动提交偏移量，得重新评估影响范围了。

@zhaogaz 目前是优先处理一下特殊情况，后续我们会排期在流量出口进行限流。流量入口的限流我们也是有计划排期优化的。感谢

@lsk569937453 这个项目我是从前辈那里接过来的。 目前已经在线上稳定运行一段时间。目前是不适宜在短时间内重构它。只想着在现有的情况下，特殊处理一下这个量大的 topic ，这个 topic 后续会下线掉。
@ZZ74 其实就是应用部署在云上有多个实例，每个实例在创建的时候都会尝试去创建 consumer 获取分区。由于都是用同一个消费者组，最终也就只有 topic 分区数的实力能够获取到该 topic 的其中一个分区，我这样理解是没问题的吧？

@ZZ74 谢谢大佬提供思路。我的想法是，只需要计算好 topic 对应的分区，多实例消费 topic 的时候，只有获取到分区的实例才能消费到数据。计算一下分区数*限流量应该就可以得到想要的结果了吧？

@wqhui 没错，就是大佬你说的意思。目前我考虑到通过使用限流器来限定特定的 consumer 1 秒只 poll 一次

@codedreamstar 不好意思之前没讲清楚，创建内存的线程池主要是为了异步处理消费者数量，consumer.poll 是不受内存线程池影响的。
@diagnostics 感谢提供思路，我之前也有考虑过用 sleep 来控制 poll 的次数，但考虑到实例多、消费 Topic 数量多等复杂情况，没有深入了解就用 sleep 感觉不太稳妥。
@lsk569937453 不好意思没有表达清楚需求。 由于存量 topic 推送过来的数据量并不大，因此目前并没有做任何限速处理，现有应用就是尽可能地去消费很多数据（依赖线程池异步处理 record)。 然后新 topic 由于数据量远远大于存量 topic 的数据量，如果不作消费限制的话，对于后续的业务处理是有着极大的压力和风险的。

感谢

gwjdYQC7EsEoimvK35Y4ZD

我也支持一下