Compartir a través de


Control.MailboxProcessor<'Msg> (Clase de F#)

Agente de procesamiento de mensajes que ejecuta un cálculo asincrónico.

Espacio de nombres/Ruta de acceso del módulo: Microsoft.FSharp.Control

Ensamblado: FSharp.Core (en 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

Comentarios

El agente encapsula una cola de mensajes que admite varios sistemas de escritura y un solo agente de lector. Los sistemas de escritura envían los mensajes al agente mediante el método Post y sus variaciones. El agente podrá esperar los mensajes mediante los métodos Receive o TryReceive, o podrá examinar todos los mensajes disponibles mediante el método Scan o TryScan.

Este tipo se denomina FSharpMailboxProcessor en el ensamblado .NET. Si obtiene acceso al tipo desde un lenguaje .NET distinto de F# o mediante reflexión, use este nombre.

Constructores

Miembro

Descripción

new

Crea un agente. La función body se usa para generar el cálculo asincrónico ejecutado por el agente. Esta función no se ejecuta hasta que se llame a Start.

Miembros de instancia

Miembro

Descripción

add_Error

Se produce cuando la ejecución del agente tiene como resultado una excepción.

CurrentQueueLength

Devuelve el número de mensajes no procesados en la cola de mensajes del agente.

DefaultTimeout

Produce una excepción de tiempo de espera si no se recibe un mensaje en este periodo de tiempo. De forma predeterminada no se usa ningún tiempo de espera.

Error

Se produce cuando la ejecución del agente tiene como resultado una excepción.

Post

Envía de forma asincrónica un mensaje a la cola de mensajes de MailboxProcessor.

PostAndAsyncReply

Envía un mensaje a un agente y espera una respuesta en el canal, de forma asincrónica.

PostAndReply

Envía un mensaje a un agente y aguarda una respuesta en el canal, de forma sincrónica.

PostAndTryAsyncReply

Igual que AsyncPostAndReply, salvo que devuelve None si no se produce ninguna respuesta dentro del periodo de tiempo de espera.

Recibir

Espera un mensaje. Se consumirá el primer mensaje por orden de llegada.

remove_Error

Se produce cuando la ejecución del agente tiene como resultado una excepción.

Scan

Busca un mensaje desplazándose por los mensajes según el orden de llegada hasta que scanner devuelve el valor Some. Los demás mensajes permanecen en la cola.

Iniciar

Inicia el agente.

TryPostAndReply

Es igual que PostAndReply, salvo en que devuelve None si no se produce ninguna respuesta dentro del plazo del tiempo de espera.

TryReceive

Espera un mensaje. Se consumirá el primer mensaje por orden de llegada.

TryScan

Busca un mensaje desplazándose por los mensajes según el orden de llegada hasta que scanner devuelve el valor Some. Los demás mensajes permanecen en la cola.

Miembros estáticos

Miembro

Descripción

Iniciar

Crea e inicia un agente. La función body se usa para generar el cálculo asincrónico ejecutado por el agente.

Ejemplo

En el siguiente ejemplo se muestra el uso básico de la clase 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

Resultados del ejemplo

  
  
  
  
  
  
  
  
  

El ejemplo siguiente muestra cómo usar MailboxProcessor para crear un agente simple que acepte varios tipos de mensajes y devuelva las respuestas adecuadas. Este agente del servidor representa a un creador de mercado, que es un agente de compra y venta en una bolsa de valores que realiza pujas y pide precios para las acciones. Los clientes pueden consultar precios, o comprar y vender acciones.

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

Resultados del ejemplo

  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  

Plataformas

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

Información de versiones

Versiones de la biblioteca básica de F#

Se admite en: 2.0, 4.0, portables

Vea también

Referencia

Microsoft.FSharp.Control (Espacio de nombres de F#)