问题标签 [mailboxprocessor]

我正在尝试创建一个字符串列表，该列表在邮箱处理器的帮助下逐渐异步插入元素。但是我没有得到想要的输出。

我几乎遵循了https://fsharpforfunandprofit.com/posts/concurrency-actor-model/中的代码， 但是在我的情况下它似乎没有按预期工作。我的代码如下：

我不断得到的结果只是最新输入的列表，不保留状态。所以如果我输入“a”然后“b”然后“c”队列将只有值“c”而不是“a”;“b”;“c”

任何帮助或指示将不胜感激！

我正在努力寻找正确的方法来处理 F# 中运行时间更长的请求/作业。

要求：

• 一个作业由多个步骤组成（需要按顺序执行）。
• 一项工作可能需要几分钟，比如说最多 10 分钟。
• 一个步骤可能涉及 IO 操作和等待时间，例如，直到由该步骤创建的文件被其他应用程序处理然后返回。
• 可能是某个步骤失败或达到了作业应提前结束的状态。
• 应该可以并行处理多个作业。
• 作业由用户请求启动/添加。
• 我希望能够根据请求跟踪作业的状态（当前步骤、先前步骤的结果）。

当前解决方案：

目前，我使用 FileSystemWatcher 来监视带有作业请求的“收件箱”。请求会导致将作业添加到由代理 (MailboxProcessor) 管理的列表中。一旦将作业添加到列表中，就会启动一个新线程（let t = new Thread(...) -> t.Start()）并且对线程的引用与作业参数一起保存（并在列表中）。在线程中，所有步骤都是按顺序执行的。这样我就可以跟踪作业状态（检查线程是否还活着）并同时处理作业。

但是，这似乎不允许我获取有关作业/线程中步骤的信息。

所需的解决方案：

另外，我想从 FileSystemWatcher 切换到基于 Suave 的 REST API。似乎我面临的问题（并行作业执行和收集有关步骤的信息，根据请求通信状态）在两个世界中都是相同的（由 FileSystemWatcher 事件或 REST API 触发的请求），但我使用 REST 方法来解释我的所需的功能：

我希望能够启动作业（POST）（响应：作业已接受，作业 ID = xyz），检查作业的状态（带有作业 ID 的 GET，包含步骤结果和当前步骤的响应）以及处理是否done 获取作业的结果（带有作业 ID 的 GET）。

至少这种设置看起来很方便，可以满足当前的需求。

任何人都可以通过向我指出处理此类要求的正确工具/方法来帮助我吗？我完全偏离了正确的方向吗？

我希望这个解释也能被我以外的其他人理解。

谢谢和最好的问候 cil

假设我调用MailboxProcessor.PostAndReply，由于任何原因，它可能会运行很长时间。如果我MailboxProcessor.Post在第一个调用尚未返回时从其他线程调用会发生什么？

我的意思是，是的，当然，我可以编写一个测试来重现这种情况。然而，在我开始重新发明自行车之前，我想知道是否有人已经知道这个问题的答案。

非常感谢！

我有一个与金融市场相连的系统，它大量使用事件。

所有代码都被构造为级联事件，中间有过滤器、聚合等。

最初该系统是用 C# 编写的，然后移植到 F#（回想起来这是一个伟大的举措），C# 代码中的事件被 F# 中的事件所取代，而没有考虑太多。

我听说过观察者模式，但我还没有真正了解过这个话题。最近，我通过一些随机浏览阅读了有关 F# 的邮箱处理器的信息。

我读到了：Difference between Observer Pattern and Event-Driven Approach，我没明白，但显然有超过 150 人投票认为答案也不太清楚 :)

在这样的文章中：https ://hackernoon.com/observer-vs-pub-sub-pattern-50d3b27f838c似乎观察者模式与事件完全相同......

乍一看，他们似乎在解决同一种问题，只是界面不同，但这让我想到了两个问题：

• 邮箱处理器真的是被使用的东西吗？它似乎主要出现在较旧的文档中，并且在我正在使用的软件包中，我没有遇到任何使用它

• 关于观察者模式，在我们使用的相当大数量的包中，只有一个包在内部使用它，但其他一切都只是使用基本事件。

是否有适合 Observable 模式和 MailboxProcessor 的特定用例？它们是否具有独特的功能？还是它们最终只是围绕事件的句法帮助？

我必须在 F# 中模拟一个离散环境，由 Python 调用，以解决强化学习问题。我有一个原始类型（主要是浮点数）的函数，可以使数据交换更顺畅。现在我可以用不同的数据多次运行这个函数，所以并行运行它似乎是个好主意。

我有以下代码：

最后我调用该函数进行测试：

我已经将它与 Array.map 进行了比较，虽然它速度更快并且使用了我笔记本电脑上 70% 的 CPU，但似乎仍然没有使用整个处理能力。我已经在具有更多内核（约 50 个）的机器上运行它，它几乎不会增加 CPU 使用率（它达到了总使用量的 3/4%，有 50 个独立输入）。我认为一定是某个地方产生了死锁，但我怎样才能检测到并摆脱它呢？

另外，为什么会这样？在我看来，函数式编程的优点之一就是能够轻松地进行并行化。

PS：SeqExtension.takeWhileWithLast 是我在 SO 上找到的一个函数，由 Tomas Petricek 在他的精彩答案之一中提供，如果需要，我可以发布它。

PPS：env是环境，其类型定义为：

我对 Async.Parallel 和 ParallelSeq 进行了同样的尝试，报告了同样的问题。

基于消息的解决方案能否解决问题>？我正在研究它，虽然我一点也不熟悉，但它会是使用 MailboxProcessor 使代码并行的好方法吗？

根据我的问题，我还尝试了这个很棒的基于数据流的并行代码库。https://nessos.github.io/Streams/。

我添加了以下代码：

我为 inputStrategy 定义了一个临时类型（基本是我拥有的旧元组），以便模拟策略只接受一个输入。不幸的是，这个问题似乎很好地隐藏在某个地方。我附上了一张关于 CPU 使用率的图表。在不同情况下在我的机器上花费的时间是：~8.8 秒（连续）；~6.2 秒（Array.Parallel.map）；~ 6.1 秒 (Nessos.Streams)

我在消息循环中有一个超时异常（我真的打算设置一个超时），我试图按如下方式捕获它

我可以看到控制台中打印的超时消息，但程序永远不会结束：我错过了什么？

这就是我printerAgent在代码中调用的方式

请注意，inbox.Receive()它最终会在几分钟后正常终止，但我的目标是设置超时（例如 30 秒）。

我已经开始阅读有关计算表达式的内容，据我所知 - 它有一些默认和自定义的隐藏实现。

我会提供我理解的东西并请纠正我。

例如，在这种情况下，我们定义了一个自定义实现来使用 let!。让每一个表情都绑定到让！记录器块内部将记录到控制台。

我记不清了，但我已经读过，如果我们不为其提供实现 - 它有一些内置的默认值。例如，当它收到 None 时，它​​会停止整个工作流并只返回 none，如果它会返回 Some - 代码将继续 -> 这是默认值吗？

由于后跟感叹号的关键字在幕后具有一些额外的功能，那么 async {} 块中的内容是什么？

举这个例子。

我认为let! msg = inbox.Receive()如果收到 None 将停止工作流。关于退货！我真的不知道。

我正在尝试制作一个正在处理订单的酒吧模拟器。发送给代理的消息属于这种类型。

代理是下面实现的一个栏类。

当我将消息发送给代理时，它正在以非常混乱的方式打印内容。

如我所见，它应该打印代码中编写的每条语句。

我已经用 F#对“经典”银行账户 kataMailboxProcessor进行了编码，以保证线程安全。但是，当我尝试将交易并行添加到帐户时，速度非常慢非常快：10 个并行调用响应（2 毫秒），20 个不响应（9 秒）！（见下面的最后一个测试Account can be updated from multiple threads）

既然MailboxProcessor支持每秒3000万条消息（见theburningmonk的文章），问题出在哪里？