移植出队 - 将循环从 C# 添加到 F#

Question

我正在尝试移植这个简单的循环：

    var messages = new List<string>();
    while (_MessageQueue.TryDequeue(out var message))
        messages.Add(message);

消息队列是并发的。它用于一个模块，该模块将来自多个线程的消息排入队列，然后由单个线程处理。

是否有一种惯用的 F# 方法来执行出列/添加循环？

Answer 1

F# 有几种实现并发的方法，包括对agent-based programming 的良好支持，因此您所做的惯用 F# 版本很可能实际上不会使用并发队列，而是基于代理或其他一些架构。

但是，要回答您关于循环的具体问题 - 由于巧妙地使用了 while 和 out var，C# 版本非常简洁。在 F# 中，您可以调用 TryDequeue 作为返回 bool 和值的方法（这样我们可以避免突变）。我会将它与递归序列表达式一起使用：

let mq = System.Collections.Concurrent.ConcurrentQueue<int>()

let rec readAll () = seq {
  let succ, msg = mq.TryDequeue() 
  if succ then
      yield msg
      yield! readAll() }

let messages = readAll() |> List.ofSeq

readAll 函数定义了一个调用TryDequeue 的序列（IEnumerable），如果操作成功，它使用yield msg 将消息添加到结果中，然后使用@987654331 递归地尝试读取更多消息@。

Answer 2

这是一个直接的转换：

    open System.Collections.Concurrent

    let _MessageQueue = ConcurrentQueue<string>()
    let messages = ResizeArray<string>()
    let mutable continueLooping = true
    while continueLooping do
        let success, message = _MessageQueue.TryDequeue()
        if success then messages.Add(message)
        continueLooping <- success

Answer 3

虽然这个问题已经有一个公认的答案，但我想使用库函数 Seq.unfold 贡献一个替代实现：

let getAllMessages (mq : _ ConcurrentQueue) =
    mq |> Seq.unfold (fun q ->
        match q.TryDequeue () with
        | true, m -> Some (m, q)
        | _ -> None)

let messages = getAllMessages _MessageQueue |> Seq.toList

不确定它在内部是否与 Tomas 的解决方案一样复杂（甚至更复杂），但我发现它简短、易于理解且优雅。

Answer 4

我为咯咯笑提供了几个额外的设计选项。常用部分：

open System.Collections.Concurrent

type Message = { I: int }

let queue = ConcurrentQueue<Message>()

drain1 调用 queue.GetEnumerator()，它的条件是它在初始请求时返回一个快照。快照基本上与 C# 版本中的竞争条件相同。

let drain1 () = queue |> Seq.toList

drain2 返回一个数组，同样是初始请求时的快照。以防万一您有幸更改返回类型。

let drain2 () = queue.ToArray()

这是 TryQueue 的惯用返回示例，它避免了“out”参数，因此它不是 C# 所做/所做的可变值。

let example () =
    let (success, message) = queue.TryDequeue()
    () // ...

最后，一个递归构建的自终止序列。

let drain3 () =

    let rec drain () = seq {
        let success, message = queue.TryDequeue()
        if success then
            yield message
            yield! drain()
        }

    drain() |> Seq.toList

（适用标准互联网保修。）

Answer 5

在@Scott Hutchinson's answer上翻拍

我会定义一个高效、直接的助手来封装突变和循环：-

[<AutoOpen>]
module ConcurrentQueueExtensions =
    type System.Collections.Concurrent.ConcurrentQueue<'T> with
        member this.Drain() =
            let buffer = ResizeArray(this.Count)
            let mutable more = true
            while more do
                match this.TryDequeue() with
                | true, req -> buffer.Add req
                | false, _ -> more <- false
            buffer.ToArray()

甚至将通用助手留在 C# 中：

class ConcurrentQueueExtensions
{
    public static T[] Drain<T>(this System.Collections.Concurrent.ConcurrentQueue<T> that)
    {
        var buffer = new List<T>(that.Count);
        while (that.TryDequeue(out var req))
            buffer.Add(req);
        return buffer.ToArray();
    }
}

然后在不混合范式的情况下应用它：

let messages = queue.Drain()