f# - 邮箱处理器，最新消息

Question

设置类似于 this .一个代理 (dataSource) 正在生成数据，而单个代理 (dataProcessor) 正在处理数据。 dataProcessor 无法处理正在生成的数据，我对处理所有消息不感兴趣，只处理最新的数据。

Jon Harrop 在那里提出了一种可能的解决方案，“当收到一条消息时贪婪地吃掉收件箱中的所有消息，并丢弃除最新消息以外的所有消息”。另一种方法是不监听所有消息，而是让dataProcessor 到PostAndReply dataSource 获取最新的一条数据。

这些方法的优缺点是什么？

Answer 1

这是一个有趣的问题，很可能有几种可能的观点。我认为最值得注意的方面是选择将影响您在两个组件之间的接口(interface)处设计 API 的方式:

1. 在“全部消费”方法中，生产者有一个非常简单的 API，只要产生一个值，它就会触发一些事件，您的消费者将订阅它。这意味着您可以让其他订阅者收听生产者的更新，并从这个问题中做一些不同于您的消费者的事情。

2. 在“调用以获取最新”方法中，可能需要编写生产者以使其保持当前状态并丢弃旧值。然后它将提供阻塞异步 API 以获取最新值。不过，它仍然可以为其他消费者公开事件。消费者将需要主动轮询更改(在某种繁忙的循环中)。

3. 您还可以让生产者拥有一个事件，如“Consume all”中那样，然后创建另一个组件来监听任何给定事件，保留最新值并通过对任何其他事件的阻塞异步调用使其可用客户。

这里有一些我能想到的优点/缺点:

• 在(1)中制作者非常简单；消费者更难写
• 在(2)中生产者需要做更多的工作，但消费者很简单
• 在 (3) 中，您将添加另一层，但是以一种相当可重用的方式。

在检查它不会影响性能后，我可能会选择 (2)(如果我只需要一个数据源)或 (3)。

至于 (3)，我的想法大概是这样的:

type KeepLastMessage<'T> = 
  | Update of 'T
  | Get of AsyncReplyChannel<'T>

type KeepLast<'T>(initial:'T, event:IObservable<'T>) = 
  let agent = MailboxProcessor.Start(fun inbox -> 
    let rec loop last = async {
      let! msg = inbox.Receive()
      match msg with 
      | Update last -> return! loop last
      | Get ch -> ch.Reply(last); return! loop last }
    loop initial)
  member x.AsyncGet() = agent.PostAndAsyncReply(Get)