Класс Control.MailboxProcessor<'Msg> (F#)
Агент обработки сообщений, выполняющий асинхронное вычисление.
Пространство имен/путь к модулю: Microsoft.FSharp.Control
Сборка: FSharp.Core (в 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
Заметки
Агент инкапсулирует очередь сообщений, которая поддерживает несколько модулей записи и один агент чтения. Модули записи отправляют сообщения агенту с помощью метода Post и его разновидностей. Агент может ожидать сообщений методом Receive или TryReceive или сканировать все доступные сообщения методом Scan или TryScan.
В сборке .NET имя этого типа — FSharpMailboxProcessor. При обращении к данному типу из языка платформы .NET Framework, отличного от F#, или посредством отражения следует использовать это имя.
Конструкторы
Элемент |
Описание |
|---|---|
Создает агент. Функция body используется для создания асинхронного вычисления, выполняемого агентом. Эта функция не выполняется до вызова метода Start. |
Члены экземпляра
Элемент |
Описание |
|---|---|
Вызывается, если в результате выполнения агента возникает исключение. |
|
Возвращает число необработанных сообщений в очереди сообщений агента. |
|
Создает исключение времени ожидания при неполучении сообщения в течение этого периода времени. По умолчанию время ожидания не используется. |
|
Вызывается, если в результате выполнения агента возникает исключение. |
|
Отправляет сообщение в очередь сообщений MailboxProcessor асинхронным образом. |
|
Отправляет агенту сообщение и ожидает ответа по каналу асинхронным образом. |
|
Отправляет агенту сообщение и одновременно ожидает ответа по каналу. |
|
Действует как AsyncPostAndReply, но возвращает None при отсутствии ответа в течение периода ожидания. |
|
Ожидает сообщение. Будет использовано первое сообщение в порядке поступления. |
|
Вызывается, если в результате выполнения агента возникает исключение. |
|
Сканирует сообщение, выполняя поиск по сообщениям в порядке поступления до тех пор, пока функция scanner не вернет значение Some. Остальные сообщения остаются в очереди. |
|
Запускает агент. |
|
Действует как PostAndReply, но возвращает None при отсутствии ответа в течение периода ожидания. |
|
Ожидает сообщение. Будет использовано первое сообщение в порядке поступления. |
|
Сканирует сообщение, выполняя поиск по сообщениям в порядке поступления до тех пор, пока функция scanner не вернет значение Some. Остальные сообщения остаются в очереди. |
Статические члены
Элемент |
Описание |
|---|---|
Создает и запускает агент. Функция body используется для создания асинхронного вычисления, выполняемого агентом. |
Пример
В следующем примере показано базовое использование класса 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
Пример результатов
В следующем примере показано как использовать MailboxProcessor для создания простого агент, который принимает различные типы сообщений и возвращает соответствующие ответы. Этот агент сервера представляет участника рынка, т. е. покупающего и продающего агента на бирже, делающего ставки и узнающего цены на активы. Клиенты могут запрашивать цены или покупать и продавать акции.
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
Пример результатов
Платформы
Windows 7, Windows Vista с пакетом обновления 2 (SP2), Windows XP с пакетом обновления 3 (SP3), Windows XP x64 с пакетом обновления 2 (SP2), Windows Server 2008 R2, Windows Server 2008 с пакетом обновления 2 (SP2), Windows Server 2003 с пакетом обновления 2 (SP2)
Сведения о версии
Среда выполнения F#
Поддерживается в версиях 2.0, 4.0
Silverlight
Поддерживается в версии 3
См. также
Ссылки
Пространство имен Microsoft.FSharp.Control (F#)
Журнал изменений
Дата |
Журнал |
Причина |
|---|---|---|
|
Август 2010 |
Добавлен пример кода. |
Улучшение информации. |