背景
在《设计事件驱动系统》一书中，作者强调了业务系统中的一个常见用例，其中需要等待或N事件发生。
给出的示例是一个订单服务，它需要等待三个单独的验证服务(均通过同一主题发送(才能返回PASS。(我认为这意味着在同一主题上会有三条具有相同密钥的验证消息，每条消息都有一个值指示成功或失败。(
作者指出，解决方案将采取以下形式(假设计数是基于密钥的(：
- 按键分组
- 统计每个键的出现次数(使用与窗口一起执行的聚合器(
- 筛选所需计数的输出
问题
上面的每一个步骤究竟是如何工作的，所涉及的类/方法是什么？
- 特别是，第一步(按密钥分组(是否指使用KStream:：groupByKey方法
- 如果是这样，输出将是一个KGroupdStream，第二步可能使用其中一个Count方法并返回一个KTable
- 然而，对于KTable，我们如何根据步骤三过滤KTable上的输出
我想你的假设是对的。对于步骤(2(，这取决于情况，但如果您假设您正在等待的所有消息都具有相同的密钥，并且只有在收到三条具有相同密钥的消息时您才感兴趣，那么调用count()就是您想要的。
作为下一步(即(3((，您将调用KTable#filter()来获取计数为3的所有行。
最后，您可以调用toStream()，并且每当一个键达到计数3时，该流都应该包含一条记录。
(附带说明：默认情况下，所有条目都将永远保留在KTable中，因此您还需要注意删除在某个时刻达到计数3的条目。(