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

Question

我是一名机械工程研究生,我的顾问刚刚让我为我们的一个传感器项目编写数据可视化实用程序.因为这是夏天,他希望我对它有一些乐趣,我认为这是学习一门擅长科学计算的语言的好时机,所以我继续前进,直接进入F#.

由于我是功能编程范例的新手,我在构建程序时遇到了一些困难,特别是考虑到在F#中轻松组合OO/FP的可能性.我的任务如下:

1. 我们有数十个传感器不断报告数据(每隔几秒钟一次).
2. 我需要同时连接所有传感器,创建每个传感器输出的内存时间序列,然后实时计算这些时间序列的各种统计数据.
3. 每隔几个小时我就需要将数据刷新到二进制文件中以进行日志记录.

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

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

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

任何建议,想法或参考表示赞赏.

谢谢 :)

Answer 1

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

但是,既然你问过,我会使用邮箱处理器作为所有传感器输入的线程安全消息处理队列.消息队列可以重新计算每个接收消息的统计信息.在我的头顶,我正在考虑这样的设置:

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,只能使用邮箱处理器实现.它可能过度,可能恰到好处,或者可能正是您的项目所需要的,这取决于您的要求.