F# MailboxProcessor极限并行

Question

我是F#新手，我尝试用MailboxProcessor进行实验，以确保状态更改是在隔离状态下完成的。

简而言之，我将操作(描述状态的不可变对象)提交给MailboxProcessor，在递归函数中，我读取消息并生成一个新状态(即向下面示例中的集合添加一个项)并将该状态发送到下一个递归。

open System

type AppliationState =
    {
        Store : string list
    }
    static member Default = 
        {
            Store = List.empty
        }
    member this.HandleAction (action:obj) =
        match action with
        | :? string as a -> { this with Store = a :: this.Store }
        | _ -> this

type Agent<'T> = MailboxProcessor<'T>     

[<AbstractClass; Sealed>]
type AppHolder private () =
    static member private Processor = Agent.Start(fun inbox ->
        let rec loop (s : AppliationState) =
            async {
                let! action = inbox.Receive()
                let s' = s.HandleAction action
                Console.WriteLine("{s: " + s.Store.Length.ToString() + " s': " + s'.Store.Length.ToString())
                return! loop s'
                }
        loop AppliationState.Default)

    static member HandleAction (action:obj) =
        AppHolder.Processor.Post action

[<EntryPoint>]
let main argv =
    AppHolder.HandleAction "a"
    AppHolder.HandleAction "b"
    AppHolder.HandleAction "c"
    AppHolder.HandleAction "d"

    Console.ReadLine()
    0 // return an integer exit code

预期产出如下：

s: 0 s': 1
s: 1 s': 2
s: 2 s': 3
s: 3 s': 4  

我得到的是：

s: 0 s': 1
s: 0 s': 1
s: 0 s': 1
s: 0 s': 1

阅读MailboxProcessor的文档并在谷歌上搜索它，我的结论是，它是一个由‘单线程’处理的消息队列，相反，它们看起来都是并行处理的。

我是不是完全离场了？

Answer 1

我认为问题一定是在HandleAction的实现中。我实现了以下步骤，它产生了预期的输出。

open System

type ApplicationState =
    {
        Items: int list
    }
    static member Default = {Items = []}
    member this.HandleAction x = {this with Items = x::this.Items}

type Message = Add of int

let Processor = MailboxProcessor<Message>.Start(fun inbox ->
    let rec loop (s : ApplicationState) =
        async {
            let! (Add action) = inbox.Receive()
            let s' = s.HandleAction action
            Console.WriteLine("s: " + s.Items.Length.ToString() + " s': " + s'.Items.Length.ToString())
            return! loop s'
        }
    loop ApplicationState.Default)

Processor.Post (Add 1)
Processor.Post (Add 2)
Processor.Post (Add 3)
Processor.Post (Add 4)


// OUTPUT
// s: 0 s': 1
// s: 1 s': 2
// s: 2 s': 3
// s: 3 s': 4

编辑

在查看了更新的代码示例之后，我相信正确的F#解决方案只是将AppHolder类型从类切换到模块。更新后的代码如下：

open System

type AppliationState =
    {
        Store : string list
    }
    static member Default = 
        {
            Store = List.empty
        }
    member this.HandleAction (action:obj) =
        match action with
        | :? string as a -> { this with Store = a :: this.Store }
        | _ -> this

type Agent<'T> = MailboxProcessor<'T>     

module AppHolder =
    let private processor = Agent.Start(fun inbox ->
        let rec loop (s : AppliationState) =
            async {
                let! action = inbox.Receive()
                let s' = s.HandleAction action
                Console.WriteLine("{s: " + s.Store.Length.ToString() + " s': " + s'.Store.Length.ToString())
                return! loop s'
            }
        loop AppliationState.Default)

    let handleAction (action:obj) =
        processor.Post action


AppHolder.handleAction "a"
AppHolder.handleAction "b"
AppHolder.handleAction "c"
AppHolder.handleAction "d"

这将产生与以前相同的结果：

{s: 0 s': 1
{s: 1 s': 2
{s: 2 s': 3
{s: 3 s': 4

Answer 2

问题是，您认为每次AppHolder.Processor都是同一个对象，但实际上每次都是不同的MailboxProcessor。我将您的AppHolder代码更改为：

[<AbstractClass; Sealed>]
type AppHolder private () =
    static member private Processor =
        printfn "Starting..."
        Agent.Start(fun inbox ->
        let rec loop (s : AppliationState) =
            async {
                let! action = inbox.Receive()
                let s' = s.HandleAction action
                printfn "{s: %A s': %A}" s s'
                return! loop s'
                }
        loop AppliationState.Default)

    static member HandleAction (action:obj) =
        AppHolder.Processor.Post action

我所做的唯一更改是简化Console.WriteLine调用，以使用printfn和%A获得更多调试细节，并添加一个printfn "Starting..."调用，该调用将在构建和启动MailboxProcessor之前立即执行。我得到的输出是：

Starting...
Starting...
Starting...
Starting...
{s: {Store = [];} s': {Store = ["b"];}}
{s: {Store = [];} s': {Store = ["d"];}}
{s: {Store = [];} s': {Store = ["c"];}}
{s: {Store = [];} s': {Store = ["a"];}}

注意，printfn "Starting..."行已经执行了四次。

这捕获了很多F#新手：member关键字定义了一个属性，而不是一个字段。每次计算属性时，都会重新计算该属性的主体。因此，每次访问AppHolder.Processor时，都会得到一个新的MailboxProcessor。有关更多详细信息，请参阅https://learn.microsoft.com/en-us/dotnet/fsharp/language-reference/members/properties。

你可能想要的是：

[<AbstractClass; Sealed>]
type AppHolder private () =
    static let processor =
        printfn "Starting..."
        Agent.Start(fun inbox ->
            // ...
        )

    static member HandleAction (action:obj) =
        processor.Post action