【问题标题】：How to schedule periodical task based on number of processed messages?如何根据处理的消息数安排定期任务？
【发布时间】：2019-01-02 01:07:40
【问题描述】：
我想使用 Kafka 处理器 API 来处理来自 Kafka 的消息。
我想调用一些周期性的函数——比如：
context.schedule(IntervalMS,punctuationType, somePunctuator)，其中 somePunctuator 执行一些定期工作，但使用间隔时间作为触发器，我想在处理一些消息后调用该任务
是否可以在 Kafka 流中进行这种触发？
【问题讨论】：
标签：
apache-kafka
apache-kafka-streams
【解决方案1】：
是的，可以使用 Kafka Streams State Store。
逻辑取决于您在达到已处理消息的数量时究竟需要做什么。
如果您需要将数据传播到下一个处理器或接收器节点，您需要将聚合值存储为键值状态存储中的对象列表。在Processor.process(..) 中，您将数据放入键值存储中，然后检查项目数是否达到限制，并执行所需的逻辑（如processorContext.forward(..)）。请看类似的例子here。
如果你在达到数字后需要做一些逻辑并且不需要值，你可以只存储计数器，并在Processor.process(..) 内部增加这个值。