我有一个用例，需要对每个传入消息应用多个函数，每个函数产生0个或更多结果。
有一个循环对我来说不会伸缩，理想的情况下，我希望能够在结果准备好后立即发出，而不是等待所有函数都被应用。
我考虑过使用AsyncIO来实现这一点，维护一个ThreadPool，但如果我没有弄错的话，我只能使用此API发出一条记录，这不会破坏交易，但我想知道是否有其他选项，例如在Map/Process函数中使用ThreadPool，以便在准备好结果时发送它们。
这是否是一个反模式，或者会导致检查点、至少一次保证方面的任何问题？
1条答案
按热度按时间vnjpjtjt1#
根据所涉及的不同函数的数量，一种解决方案是将每个传入消息扇出到 n 个操作符，每个操作符应用一个函数。
我担心，如果您尝试使用多线程Map/处理函数，您会遇到麻烦。
不如这样吧：
您可以使用RichCoFlatMap之类的工具（或KeyedCoProcessFunction，或BroadcastProcessFunction），它知道所有当前活动的函数，并且对于每个传入事件，发出它的 n 个副本，每个副本用关于要执行的特定函数的信息来丰富。它负责执行函数并在结果可用时发出结果。