Share via


Classe Control.MailboxProcessor<'Msg> (F#)

Agente di elaborazione dei messaggi che esegue un calcolo asincrono.

Percorso spazio dei nomi/modulo: Microsoft.FSharp.Control

Assembly: FSharp.Core (in FSharp.Core.dll)

[<Sealed>]
[<AutoSerializable(false)>]
type MailboxProcessor<'Msg> =
 class
  interface IDisposable
  new MailboxProcessor : (MailboxProcessor<'Msg> -> Async<unit>) * ?CancellationToken -> MailboxProcessor<'Msg>
  member this.Post : 'Msg -> unit
  member this.PostAndAsyncReply : (AsyncReplyChannel<'Reply> -> 'Msg) * int option -> Async<'Reply>
  member this.PostAndReply : (AsyncReplyChannel<'Reply> -> 'Msg) * int option -> 'Reply
  member this.PostAndTryAsyncReply : (AsyncReplyChannel<'Reply> -> 'Msg) * ?int -> Async<'Reply option>
  member this.Receive : ?int -> Async<'Msg>
  member this.Scan : ('Msg -> Async<'T> option) * ?int -> Async<'T>
  member this.Start : unit -> unit
  static member Start : (MailboxProcessor<'Msg> -> Async<unit>) * ?CancellationToken -> MailboxProcessor<'Msg>
  member this.TryPostAndReply : (AsyncReplyChannel<'Reply> -> 'Msg) * ?int -> 'Reply option
  member this.TryReceive : ?int -> Async<'Msg option>
  member this.TryScan : ('Msg -> Async<'T> option) * ?int -> Async<'T option>
  member this.add_Error : Handler<Exception> -> unit
  member this.CurrentQueueLength :  int
  member this.DefaultTimeout :  int with get, set
  member this.Error :  IEvent<Exception>
  member this.remove_Error : Handler<Exception> -> unit
 end

Note

L'agente incapsula una coda di messaggi che supporta più writer e un solo agente di lettura.I writer inviano messaggi all'agente tramite il metodo Post e le relative varianti.L'agente può attendere i messaggi utilizzando i metodi Receive o TryReceive o analizzare tutti i messaggi disponibili con il metodo Scan o TryScan.

Questo tipo è denominato FSharpMailboxProcessor nell'assembly .NET.Utilizzare questo nome per accedere al tipo da un linguaggio .NET diverso da F# o tramite reflection.

Costruttori

Membro

Descrizione

new

Crea un agente.La funzione body viene utilizzata per generare il calcolo asincrono eseguito dall'agente.Questa funzione non viene eseguita finché non viene chiamato Start.

Membri di istanza

Membro

Descrizione

add_Error

Si verifica quando l'esecuzione dell'agente genera un'eccezione.

CurrentQueueLength

Restituisce il numero di messaggi non elaborati nella coda di messaggi dell'agente.

DefaultTimeout

Genera un'eccezione di timeout se non è stato ricevuto un messaggio nell'arco di tempo specificato.Per impostazione predefinita, non viene utilizzato alcun timeout.

delle modifiche a..."

Si verifica quando l'esecuzione dell'agente genera un'eccezione.

Post

Inserisce un messaggio nella coda dei messaggi di MailboxProcessor in modo asincrono.

PostAndAsyncReply

Invia un messaggio a un agente e attende una risposta sul canale in modo asincrono.

PostAndReply

Invia un messaggio a un agente e attende una risposta sul canale in modo sincrono.

PostAndTryAsyncReply

Analogo a AsyncPostAndReply, ma restituisce None se non viene ricevuta alcuna risposta entro il periodo di timeout.

Receive

Attende un messaggio.Il metodo utilizzerà il primo messaggio in ordine di arrivo.

remove_Error

Si verifica quando l'esecuzione dell'agente genera un'eccezione.

Scan

Analizza un messaggio eseguendo una ricerca nei messaggi in ordine di arrivo finché scanner non restituisce un valore Some.Gli altri messaggi rimangono nella coda.

Start

Avvia l'agente.

TryPostAndReply

Come PostAndReply, ma restituisce None se non si verifica alcuna replica nel periodo di timeout.

TryReceive

Attende un messaggio.Il metodo utilizzerà il primo messaggio in ordine di arrivo.

TryScan

Analizza un messaggio eseguendo una ricerca nei messaggi in ordine di arrivo finché scanner non restituisce un valore Some.Gli altri messaggi rimangono nella coda.

Membri statici

Membro

Descrizione

Start

Crea e avvia un agente.La funzione body viene utilizzata per generare il calcolo asincrono eseguito dall'agente.

Esempio

Nell'esempio riportato di seguito vengono illustrate le basi dell'uso della classe MailboxProcessor.

open System
open Microsoft.FSharp.Control

type Message(id, contents) =
    static let mutable count = 0
    member this.ID = id
    member this.Contents = contents
    static member CreateMessage(contents) =
        count <- count + 1
        Message(count, contents)

let mailbox = new MailboxProcessor<Message>(fun inbox ->
    let rec loop count =
        async { printfn "Message count = %d. Waiting for next message." count
                let! msg = inbox.Receive()
                printfn "Message received. ID: %d Contents: %s" msg.ID msg.Contents
                return! loop( count + 1) }
    loop 0)

mailbox.Start()

mailbox.Post(Message.CreateMessage("ABC"))
mailbox.Post(Message.CreateMessage("XYZ"))


Console.WriteLine("Press any key...")
Console.ReadLine() |> ignore

Esempio di output

  
  
  
  
  
  
  
  
  

Nell'esempio riportato di seguito viene illustrato come utilizzare MailboxProcessor per creare un agente semplice che accetti vari tipi di messaggio e restituisca le risposte corrette.Questo agente del server rappresenta un market maker, ovvero un agente di vendita e di acquisto di borsa che imposta l'offerta e chiede i prezzi delle risorse.I client possono eseguire una query per i prezzi o comprare e vendere azioni.

open System

type AssetCode = string

type Asset(code, bid, ask, initialQuantity) =
    let mutable quantity = initialQuantity
    member this.AssetCode = code
    member this.Bid = bid
    member this.Ask = ask
    member this.Quantity with get() = quantity and set(value) = quantity <- value


type OrderType =
    | Buy of AssetCode * int
    | Sell of AssetCode * int

type Message =
    | Query of AssetCode * AsyncReplyChannel<Reply>
    | Order of OrderType * AsyncReplyChannel<Reply>
and Reply =
    | Failure of string
    | Info of Asset
    | Notify of OrderType

let assets = [| new Asset("AAA", 10.0, 10.05, 1000000);
                new Asset("BBB", 20.0, 20.10, 1000000);
                new Asset("CCC", 30.0, 30.15, 1000000) |]

let codeAssetMap = assets
                   |> Array.map (fun asset -> (asset.AssetCode, asset))
                   |> Map.ofArray

let mutable totalCash = 00.00
let minCash = -1000000000.0
let maxTransaction = 1000000.0

let marketMaker = new MailboxProcessor<Message>(fun inbox ->
    let rec Loop() =
        async {
            let! message = inbox.Receive()
            match message with
            | Query(assetCode, replyChannel) ->
                match (Map.tryFind assetCode codeAssetMap) with
                | Some asset ->
                    printfn "Replying with Info for %s" (asset.AssetCode)
                    replyChannel.Reply(Info(asset))
                | None -> replyChannel.Reply(Failure("Asset code not found."))
            | Order(order, replyChannel) ->
                match order with
                | Buy(assetCode, quantity) ->
                    match (Map.tryFind assetCode codeAssetMap) with
                    | Some asset ->
                        if (quantity < asset.Quantity) then
                            asset.Quantity <- asset.Quantity - quantity
                            totalCash <- totalCash + float quantity * asset.Ask
                            printfn "Replying with Notification:\nBought %d units of %s at price $%f. Total purchase $%f."
                                    quantity asset.AssetCode asset.Ask (asset.Ask * float quantity)
                            printfn "Marketmaker balance: $%10.2f" totalCash
                            replyChannel.Reply(Notify(Buy(asset.AssetCode, quantity)))
                        else
                            printfn "Insufficient shares to fulfill order for %d units of %s."
                                    quantity asset.AssetCode
                            replyChannel.Reply(Failure("Insufficient shares to fulfill order."))
                    | None -> replyChannel.Reply(Failure("Asset code not found."))
                | Sell(assetCode, quantity) ->
                    match (Map.tryFind assetCode codeAssetMap) with
                    | Some asset ->
                        if (float quantity * asset.Bid <= maxTransaction && totalCash - float quantity * asset.Bid > minCash) then
                            asset.Quantity <- asset.Quantity + quantity
                            totalCash <- totalCash - float quantity * asset.Bid
                            printfn "Replying with Notification:\nSold %d units of %s at price $%f. Total sale $%f."
                                    quantity asset.AssetCode asset.Bid (asset.Bid * float quantity)
                            printfn "Marketmaker balance: $%10.2f" totalCash
                            replyChannel.Reply(Notify(Sell(asset.AssetCode, quantity)))
                        else
                            printfn "Insufficient cash to fulfill order for %d units of %s."
                                    quantity asset.AssetCode
                            replyChannel.Reply(Failure("Insufficient cash to cover order."))
                    | None -> replyChannel.Reply(Failure("Asset code not found."))
            do! Loop()
        }
    Loop())

marketMaker.Start()

// Query price.
let reply1 = marketMaker.PostAndReply(fun replyChannel -> 
    printfn "Posting message for AAA"
    Query("AAA", replyChannel))

// Test Buy Order.
let reply2 = marketMaker.PostAndReply(fun replyChannel -> 
    printfn "Posting message for BBB"
    Order(Buy("BBB", 100), replyChannel))

// Test Sell Order.
let reply3 = marketMaker.PostAndReply(fun replyChannel -> 
    printfn "Posting message for CCC"
    Order(Sell("CCC", 100), replyChannel))

// Test incorrect code.
let reply4 = marketMaker.PostAndReply(fun replyChannel -> 
    printfn "Posting message for WrongCode"
    Order(Buy("WrongCode", 100), replyChannel))

// Test too large a number of shares.

let reply5 = marketMaker.PostAndReply(fun replyChannel ->
    printfn "Posting message with large number of shares of AAA."
    Order(Buy("AAA", 1000000000), replyChannel))

// Too large an amount of money for one transaction.

let reply6 = marketMaker.PostAndReply(fun replyChannel ->
    printfn "Posting message with too large of a monetary amount."
    Order(Sell("AAA", 100000000), replyChannel))

let random = new Random()
let nextTransaction() =
    let buyOrSell = random.Next(2)
    let asset = assets.[random.Next(3)]
    let quantity = Array.init 3 (fun _ -> random.Next(1000)) |> Array.sum
    match buyOrSell with
    | n when n % 2 = 0 -> Buy(asset.AssetCode, quantity)
    | _ -> Sell(asset.AssetCode, quantity)

let simulateOne() =
   async {
       let! reply = marketMaker.PostAndAsyncReply(fun replyChannel ->
           let transaction = nextTransaction()
           match transaction with
           | Buy(assetCode, quantity) -> printfn "Posting BUY %s %d." assetCode quantity
           | Sell(assetCode, quantity) -> printfn "Posting SELL %s %d." assetCode quantity
           Order(transaction, replyChannel))
       printfn "%s" (reply.ToString())
    }

let simulate =
    async {
        while (true) do
            do! simulateOne()
            // Insert a delay so that you can see the results more easily.
            do! Async.Sleep(1000)
    }

Async.Start(simulate)

Console.WriteLine("Press any key...")
Console.ReadLine() |> ignore

Esempio di output

  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  

Piattaforme

Windows 8, Windows 7, Windows Server 2012, Windows Server 2008 R2

Informazioni sulla versione

Versioni della libreria di base F#

Supportato in: 2,0, 4,0, portabile

Vedere anche

Riferimenti

Spazio dei nomi Microsoft.FSharp.Control (F#)