asynchronous - 使用 LIFO 逻辑运行的 MailboxProcessor

Question

我正在学习 F# 代理 (MailboxProcessor)。

我正在处理一个相当不寻常的问题。

• 我有一个代理 (dataSource)，它是流数据源。数据必须由一组代理 (dataProcessor) 进行处理。我们可以将 dataProcessor 视为某种跟踪设备。
• 数据流入的速度可能快于dataProcessor处理其输入的速度。
• 有一些延迟是可以接受的。但是，我必须确保代理始终掌控其工作，并且不会堆积在过时的观察结果中

我正在探索解决这个问题的方法。

第一个想法是实现 stack (后进先出)在数据源中。当 dataProcessor 可用于接收和处理数据时，dataSource 将发送可用的最新观察结果。此解决方案可能有效，但可能会变得复杂，因为 dataProcessor 可能需要被阻止并重新激活；并将其状态传达给dataSource，从而导致双向通信问题。此问题可能归结为 consumer-producer problem 中的阻塞队列。但我不确定..

第二个想法是让dataProcessor负责消息排序。在此架构中，dataSource 将简单地将更新发布到 dataProcessor 的队列中。 dataProcessor 将使用 Scan 获取队列中可用的最新数据。这可能是要走的路。但是，我不确定在 MailboxProcessor 的当前设计中是否可以清除消息队列，删除较旧的过时消息。此外，here ，其中写道:

Unfortunately, the TryScan function in the current version of F# is broken in two ways. Firstly, the whole point is to specify a timeout but the implementation does not actually honor it. Specifically, irrelevant messages reset the timer. Secondly, as with the other Scan function, the message queue is examined under a lock that prevents any other threads from posting for the duration of the scan, which can be an arbitrarily long time. Consequently, the TryScan function itself tends to lock-up concurrent systems and can even introduce deadlocks because the caller's code is evaluated inside the lock (e.g. posting from the function argument to Scan or TryScan can deadlock the agent when the code under the lock blocks waiting to acquire the lock it is already under).

最新观察结果的反弹可能是一个问题。这篇文章的作者 @Jon Harrop 建议

I managed to architect around it and the resulting architecture was actually better. In essence, I eagerly Receive all messages and filter using my own local queue.

这个想法确实值得探索，但在开始使用代码之前，我欢迎一些关于如何构建我的解决方案的意见。

谢谢。

Answer 1

听起来您可能需要邮箱处理器的破坏性扫描版本，我在您可能感兴趣的博客系列中使用 TPL Dataflow 实现了此功能。

我的博客目前已关闭以进行维护，但我可以向您指出 Markdown 格式的帖子。

Part1
Part2
Part3

您还可以查看 github 上的代码

我还写了关于我潜伏的恐惧中的扫描问题post

希望有帮助...