如何用Streamly消费无限单子流并控制并发HTTP调用数？

这个问题我之前也踩过类似的坑，Streamly异步组合子的预取机制确实容易让人困惑，不过调整一下代码结构就能完美满足你的需求：

问题根源 #

你原来的写法S.take 10 $ (asyncly getStream)之所以会发起超过10个HTTP调用，是因为asyncly为了提升吞吐量，会提前从上游的无限流预取元素——也就是说，在你拿到前10个结果之前，它已经悄悄发起了更多的HTTP请求。

核心解决方案：调整流的处理顺序 #

关键思路是：先限制流的任务数量为10，再让这10个任务并发执行，从根源上避免多余的请求。

修改后的代码示例如下：

-- 你的getStream定义保持不变
getStream = foldMap makeHttpCall [1..] where 
    makeHttpCall bookId = ... -- 这里是你的IO类型HTTP调用动作

-- 正确的使用方式：先取10个任务，再异步执行
results <- S.toList $ asyncly $ S.take 10 getStream

这样做完全符合你的两个需求：

• 无需把10传入getStream函数，保持了函数的通用性
• S.take 10 getStream会从无限流中精确取出前10个HTTP调用任务，asyncly会异步执行这10个任务，确保同时有10个HTTP请求在运行

额外选项：控制结果顺序 #

如果你希望结果按原始请求的顺序返回（而不是任务完成的先后顺序），只需要把asyncly换成parallely即可：

results <- S.toList $ parallely $ S.take 10 getStream

效果验证 #

这个写法在Streamly 0.2.0中是完全有效的：上游流只会生成10个任务，异步/并行组合子只会执行这10个任务，不会有多余的HTTP请求发起。

内容的提问来源于stack exchange，提问作者zoran119