f# - 将异步调用限制为每分钟最多 n

Question

假设我有一个 async这会影响一些外部服务:

fetchFoo : Async<string>

为了不至于对服务造成太大影响，我想将其限速为 n每分钟请求数。

let fetchFooWithRateLimit : Async<string> = applyRateLimit 6 fetchFoo

如 fetchFooWithRateLimit运行超过 n每分钟的次数，它会在内部等待一点，以延迟对 fetchFoo 的底层调用。 .

我怎样才能在 F# 中实现这一点？

Answer 1

关于这个问题需要注意的一点是，在经典的生产者-消费者场景中，生产者可能会超过消费者，排队是不可避免的。

一个简单的方法是计算下一项的延迟值:如果达到速率限制，则延迟到下一个时隙。缺点是它可能最终使用线程池作为其队列。

话虽如此，我们可以使用 MailboxProcessor作为我们的异步队列实现，因为它提供了我们想要的很多开箱即用的东西。

let rateLimit fetch period limit = 
    let now () = DateTimeOffset.Now
    let cts = new CancellationTokenSource()
    let mailbox = 
        MailboxProcessor.Start(fun inbox -> 
            let rec loop nextTime remaining = async {
                let diff = int (nextTime - now()).TotalMilliseconds
                if remaining = 0 || diff < 0 then
                    do! Async.Sleep (max diff 0)
                    return! loop (now() + period) limit
                else
                    let! request = inbox.Receive()
                    do! fetch request                
                    return! loop nextTime (remaining - 1)                                        
            }

            loop (now ()) 0
        , cts.Token)

    {| Post = mailbox.Post; Stop = cts.Cancel |} 

基本思想是如果我们已经超过了速率限制，则延迟出队。

测试:

let fetch args = async { do printfn "%A %A" DateTime.Now args }

let rl = rateLimit fetch (TimeSpan.FromSeconds 5.0) 5

Observable.interval(TimeSpan.FromSeconds 0.5) |> Observable.subscribe(rl.Post)

输出:

6/4/2020 12:48:19 AM 0L
6/4/2020 12:48:19 AM 1L
6/4/2020 12:48:19 AM 2L
6/4/2020 12:48:19 AM 3L
6/4/2020 12:48:19 AM 4L
6/4/2020 12:48:23 AM 5L
6/4/2020 12:48:23 AM 6L
6/4/2020 12:48:23 AM 7L
6/4/2020 12:48:23 AM 8L
6/4/2020 12:48:23 AM 9L
6/4/2020 12:48:28 AM 10L
6/4/2020 12:48:28 AM 11L
6/4/2020 12:48:28 AM 12L
6/4/2020 12:48:28 AM 13L
6/4/2020 12:48:28 AM 14L

注意:我使用匿名记录创建了一个简单的 API Post和 Stop方法。
如果您的 F# 版本尚不支持此功能，只需将其更改为返回一个元组即可。