.net - 你会如何在 F# 中解决这个问题？ (高频传感器数据)

Question

我是一名机械工程研究生，我的顾问刚刚让我为我们的一个传感器项目编写一个数据可视化实用程序。由于现在是夏天，他希望我能从中获得一些乐趣，我认为这将是学习一门擅长科学计算的语言的好时机，所以我继续努力并直接进入 F#。

由于我是函数式编程范式的新手，因此我在正确构建我的程序时遇到了一些困难，尤其是考虑到在 F# 中轻松组合 OO/FP 的可能性。我的任务如下:

我们有数十个传感器不断报告数据(每隔几秒一次)。

我需要同时连接到所有传感器，创建每个传感器输出的内存时间序列，然后实时计算这些时间序列的各种统计数据。

每隔几个小时，我需要将数据刷新到二进制文件中以进行记录。

我应该如何设计我的应用程序？我想过这样的事情:
1. 我计划连接到每个传感器以开始接收数据，然后将这些数据转储到消息队列中。
2. 我有一个事件驱动的处理函数来接收队列上的数据。接收到数据后，判断数据来自哪个传感器，然后将数据放入对应传感器的timeseries对象中。
3. 每次添加传感器数据时间序列对象时，我都可以触发一个事件并让我的统计函数为传感器处理新数据。

显然我需要在这个应用程序中保持某种状态。所以我会添加以下可变数据结构。我将使用通用的 .NET 可调整大小的列表来存储我的时间序列数据并实现一个新的派生类来触发数据添加事件。我可以将sensorid 和实际时间序列容器之间的映射存储在字典中(当数据从队列中弹出时，我可以读取sensorid 字段，获取该sensorid 的时间序列容器，然后轻松添加新数据)。我还可以有第二个字典来存储sensorid 和包含该sensorid 时间序列统计信息的各种时间序列之间的映射)。当一个主传感器时间序列被添加到时，它会触发一个事件来调用所有的统计函数来在新数据上运行自己，并将它们的信息存储在该传感器 ID 的适当字典中。

我还没有想过如何保存数据，但我想我可以用数据写出二进制文件。

感谢任何建议、想法或引用。

谢谢 :)

Answer 1

我建议不要在 F# 中实现你的新项目，直到你更好地掌握语言，否则你最终会用 F# 语法编写 C# 代码。至少对于工作中的项目而言，使用您熟悉的工具比使用公司资金来试验新技术更好。

但是，既然你问了，我会用 mailbox processor充当所有传感器输入的线程安全消息处理队列。消息队列可以重新计算接收到的每条消息的统计信息。在我的头顶上，我正在考虑这样的设置:

type SensorMsg<'a, 'b> =
    | Fetch of 'a AsyncReplyChannel
    | Post of 'b
    | Die

type SensorMessageQueue<'a, 'b>(emptyStats : 'a, compute : 'a -> 'b -> 'a) =
    let queue = MailboxProcessor.Start(fun inbox ->
            let rec loop stats =
                async {
                    let! msg = inbox.Receive()
                    match msg with
                    | Die -> return ()
                    | Post(x) -> return! loop (compute stats x)
                    | Fetch(x) -> x.Reply(stats); return! loop stats
                }
            loop emptyStats
        )

    member this.Post(x) = queue.Post(Post(x))
    member this.Fetch() = queue.PostAndReply(fun replyChannel -> Fetch(replyChannel))
    member this.Die() = queue.Post(Die)

像这样的东西可以保存传感器的实时运行统计信息。例如，假设我想发布一些可以保持运行平均值的内容:

let averager =
    SensorMessageQueue(
            (0, 0), (* initial state of sum, total *)
            (fun (sum, total) input -> sum + input, total + 1)
        )

averager.Post(75)
averager.Post(90)
averager.Post(80)
let x = averager.Fetch() (* returns (245, 3) *)
averager.Post(100)
let y = averager.Fetch() (* returns (345, 4) *)

像这样的设置相对容易使用，线程安全，并且不使用可变状态(所有“状态”都存在于闭包的参数中)。仔细想想，这基本上是一个美化的 seq.unfold，仅使用邮箱处理器实现。它可能是矫枉过正，可能恰到好处，或者可能正是您的项目所需要的，这取决于您的要求。