基于片段和答案,是否可以将结果从限制队列返回给调用者?我已经尝试过PostAndAsyncReply在频道上接收回复,但如果我使用 Enqueue 管道它会引发错误。这是代码。
欣赏围绕队列或邮箱设计模式的基于 F# 核心原版的解决方案。
问题
问题是能够根据油门(一次最多 3 个)异步调用函数,从数组中传递每个项目,等待整个队列/数组直到它完成,同时收集所有结果,然后将结果返回给呼叫者,召集者。(将结果返回给调用者是这里待处理的内容)
被叫代码
// Message type used by the agent - contains queueing
// of work items and notification of completion
type ThrottlingAgentMessage =
| Completed
| Enqueue of Async<unit>
/// Represents an agent that runs operations in concurrently. When the number
/// of concurrent operations exceeds 'limit', they are queued and processed later
let throttlingAgent limit =
MailboxProcessor.Start(fun inbox ->
async {
// The agent body is not executing in parallel,
// so we can safely use mutable queue & counter
let queue = System.Collections.Generic.Queue<Async<unit>>()
let running = ref 0
while true do
// Enqueue new work items or decrement the counter
// of how many tasks are running in the background
let! msg = inbox.Receive()
match msg with
| Completed -> decr running
| Enqueue w -> queue.Enqueue(w)
// If we have less than limit & there is some work to
// do, then start the work in the background!
while running.Value < limit && queue.Count > 0 do
let work = queue.Dequeue()
incr running
do! // When the work completes, send 'Completed'
// back to the agent to free a slot
async {
do! work
inbox.Post(Completed)
}
|> Async.StartChild
|> Async.Ignore
})
let requestDetailAsync (url: string) : Async<Result<string, Error>> =
async {
Console.WriteLine ("Simulating request " + url)
try
do! Async.Sleep(1000) // let's say each request takes about a second
return Ok (url + ":body...")
with :? WebException as e ->
return Error {Code = "500"; Message = "Internal Server Error"; Status = HttpStatusCode.InternalServerError}
}
let requestMasterAsync() : Async<Result<System.Collections.Concurrent.ConcurrentBag<_>, Error>> =
async {
let urls = [|
"http://www.example.com/1";
"http://www.example.com/2";
"http://www.example.com/3";
"http://www.example.com/4";
"http://www.example.com/5";
"http://www.example.com/6";
"http://www.example.com/7";
"http://www.example.com/8";
"http://www.example.com/9";
"http://www.example.com/10";
|]
let results = System.Collections.Concurrent.ConcurrentBag<_>()
let agent = throttlingAgent 3
for url in urls do
async {
let! res = requestDetailAsync url
results.Add res
}
|> Enqueue
|> agent.Post
return Ok results
}
来电代码
[<TestMethod>]
member this.TestRequestMasterAsync() =
match Entity.requestMasterAsync() |> Async.RunSynchronously with
| Ok result -> Console.WriteLine result
| Error error -> Console.WriteLine error