How to dequeue when new item in queue

Question

I've an application that works with a queue with strings (which corresponds to different tasks that application needs to perform). At random moments the queue can be filled with strings (like several times a minute sometimes but it also can take a few hours.

Till now I always had a timer that checked every few seconds the queue whether there were items in the queue and removed them.

I think there must be a nicer solution than this way. Is there any way to get an event or so when an item is added to the queue?

http://stackoverflow.com/questions/3127136/observable-stack-and-queue — Habib, Jun 05 '14 at 16:02

spender · Accepted Answer · 2014-06-05T16:26:58.590

Yes. Take a look at TPL Dataflow, in particular, the BufferBlock<T>, which does more or less the same as BlockingCollection without the nasty side-effect of jamming up your threads by leveraging async/await.

So you can:

void Main()
{
    var b = new BufferBlock<string>();
    AddToBlockAsync(b);
    ReadFromBlockAsync(b);
}

public async Task AddToBlockAsync(BufferBlock<string> b)
{
    while (true)
    {
        b.Post("hello");
        await Task.Delay(1000);
    }
}

public async Task ReadFromBlockAsync(BufferBlock<string> b)
{
    await Task.Delay(10000); //let some messages buffer up...
    while(true)
    {
        var msg = await b.ReceiveAsync();
        Console.WriteLine(msg);
    }
}

Daniel Kelley · Answer 2 · 2014-06-05T16:09:44.323

2

I'd take a look at BlockingCollection.GetConsumingEnumerable. The collection will be backed with a queue by default, and it is a nice way to automatically take values from the queue as they are added using a simple foreach loop.

There is also an overload that allows you to supply a CancellationToken meaning you can cleanly break out.

edited Jun 05 '14 at 16:09

answered Jun 05 '14 at 16:03

Daniel Kelley

7,579
6
42
50

score 1 · Answer 3 · answered Jun 05 '14 at 16:08

Have you looked at BlockingCollection ? The GetConsumingEnumerable() method allows an indefinite loop to be run on the consumer, to which will new items will be yielded once an item becomes available, with no need for timers, or Thread.Sleep's:

// Common:

  BlockingCollection<string> _blockingCollection =
                     new BlockingCollection<string>();

// Producer

 for (var i = 0; i < 100; i++) 
 {
    _blockingCollection.Add(i.ToString());
    Thread.Sleep(500); // So you can track the consumer synchronization. Remove.
 }

// Consumer:

   foreach (var item in _blockingCollection.GetConsumingEnumerable())
   {
      Debug.WriteLine(item);
   }

How to dequeue when new item in queue

3 Answers3