2

尝试运行此处给出的阻塞队列示例。

使用该示例中所示的 .NET Queue 类，我收到以下错误：

let queue = new Queue<_>()

“非泛型类型 System.Collection.Queue 不需要任何类型参数”

如果从 Queue 创建中删除了 type 参数，那么type BlockingQueueAgent<'T>(maxLength).

除了实现我自己的队列类之外，还有什么方法可以使用示例程序中使用的 .NET 队列类？下面给出了整个代码，以防更具说明性。

open System
open System.IO
open System.Collections
//open System.Runtime.Serialization.Formatters.Binary

///defining Agent
type Agent<'T> = MailboxProcessor<'T>

///defining Message
type internal BlockingAgentMessage<'T> = 
  | Get of AsyncReplyChannel<'T>
  | Add of 'T * AsyncReplyChannel<unit> 

/// Agent-based implementation of producer/consumer problem 
type BlockingQueueAgent<'T>(maxLength) =
    let agent = Agent.Start(fun agent ->
        let queue = new Queue<_>()
        //let queue = new Queue()
        // State machine running inside the agent
        let rec emptyQueue() = 
            agent.Scan(fun msg ->
                match msg with 
                | Add(value, reply) -> Some(enqueueAndContinue(value, reply))
                | _ -> None )
        and fullQueue() = 
            agent.Scan(fun msg ->
                match msg with 
                | Get(reply) -> Some(dequeueAndContinue(reply))
                | _ -> None )
        and runningQueue() = async {
            let! msg = agent.Receive() 
            match msg with 
            | Add(value, reply) -> return! enqueueAndContinue(value, reply)
            | Get(reply) -> return! dequeueAndContinue(reply) }
        and enqueueAndContinue (value, reply) = async {
            queue.Enqueue(value)
            reply.Reply() 
            return! chooseState() }
        and dequeueAndContinue (reply) = async { 
            reply.Reply(queue.Dequeue())
            return! chooseState() }
        and chooseState() = 
            if queue.Count = 0 then emptyQueue()
            elif queue.Count = maxLength then fullQueue()
            else runningQueue()

        // Start with an empty queue
        emptyQueue() )

    /// Asynchronously adds item to the queue. If the queue
    /// is full, it blocks until some items are removed.
    member x.AsyncAdd(v:'T) = 
        agent.PostAndAsyncReply(fun ch -> Add(v, ch))

    /// Asynchronously gets item from the queue. If the queue
    /// is empty, it blocks until some items are added.
    member x.AsyncGet() = 
        agent.PostAndAsyncReply(Get)




let ag = new BlockingQueueAgent<int>(3)

let writer() = async { 
    for i in 0 .. 10 do 
        do! ag.AsyncAdd(i)
        printfn "Added: %d" i }

let reader () = async { 
    while true do
        let! v = ag.AsyncGet()
        do! Async.Sleep(1000)
        printfn "Got: %d" v }

reader () |> Async.Start
writer () |> Async.Start
4

1 回答 1

9

我认为您正在寻找System.Collections.Generic.Queue而不是System.Collections.Queue。

改变

open System.Collections

到

open System.Collections.Generic
于 2012-04-21T02:15:24.320 回答