Pulling from a queue using multiple tasks

[nbrege]

I need some help trying to accomplish something. I have a concurrentqueue defined in my program. Items get added to the queue in a sporadic fashion. It could be empty for a couple hours, then get 50-60 items added to it within seconds. I want to have multiple tasks (5 for example) that "watch" the queue for when items are added, then dequeue the items & perform a process on each one. So while the queue is empty, the tasks are idle. But if 20 items get added then the tasks "see" that there items in the queue & they dequeue & process the items, 5 at a time, until all 20 are gone. Then the tasks are idle again until items are added the the queue again. This gets repeated over & over, while the program is running. I'm not sure how to set this up so that the tasks start up when items are added. Do I need to use a loop to constantly be checking if the queue has items? That seems inefficient to me. Any help would be appreciated. Thanks...

[dbasnett]

Maybe this will give you some ideas,

Public Class Form1
    Private IdleTime As Integer = 250 'how long to wait between Q checks
    Private IdleCheck As New Threading.ManualResetEvent(False)
    Private MyQ As New Concurrent.ConcurrentQueue(Of String)
    Private MyTasks As New List(Of Task)

    Private Sub Form1_Load(sender As Object, e As EventArgs) Handles MyBase.Load
        For x As Integer = 1 To 5 'start five tasks
            Dim tsk As Task = Task.Run(Sub() ProcMyQ())
            MyTasks.Add(tsk)
        Next
    End Sub

    Private Sub ProcMyQ()
        Do While Not IdleCheck.WaitOne(IdleTime)
            Do While MyQ.Count > 0
                Dim s As String = ""
                If MyQ.TryDequeue(s) Then
                    'process item
                    Debug.WriteLine(s)
                End If
            Loop
        Loop
    End Sub

    Private Sub Button1_Click(sender As Object, e As EventArgs) Handles Button1.Click
        For x As Integer = 1 To 100
            MyQ.Enqueue(x.ToString)
        Next
    End Sub

    Private Sub Button2_Click(sender As Object, e As EventArgs) Handles Button2.Click
        IdleCheck.Set() 'stop the tasks
    End Sub
End Class

[PlausiblyDamp]

Would something like https://learn.microsoft.com/en-us/do...currentqueue-1 do what you are after?

[Napstore]

You don’t have to keep looping through the queue, which can waste resources. Instead, I recommend using a BlockingCollection instead of ConcurrentQueue. It’s super handy because it blocks the tasks when there are no items and wakes them up when something gets added. You can set up multiple tasks that listen for items in the collection, and when an item is added, they just grab it and process it. I’ve used this approach in a similar situation, and it worked great for managing bursts of data without keeping everything in a constant check. Just add items to the BlockingCollection, and your tasks will handle the rest.

[nbrege]

You don’t have to keep looping through the queue, which can waste resources. Instead, I recommend using a BlockingCollection instead of ConcurrentQueue. It’s super handy because it blocks the tasks when there are no items and wakes them up when something gets added. You can set up multiple tasks that listen for items in the collection, and when an item is added, they just grab it and process it. I’ve used this approach in a similar situation, and it worked great for managing bursts of data without keeping everything in a constant check. Just add items to the BlockingCollection, and your tasks will handle the rest.

Thanks, that sounds exactly like what I need. Any chance you can post a quick example of how to set it up?

[dbasnett]

Thanks, that sounds exactly like what I need. Any chance you can post a quick example of how to set it up?

Public Class Form1
    Private MyBC As New Concurrent.BlockingCollection(Of String)
    Private MyTasks As New List(Of Task)

    Private Sub Form1_FormClosing(sender As Object, e As FormClosingEventArgs) Handles Me.FormClosing
        MyBC.CompleteAdding()
    End Sub

    Private Sub Form1_Load(sender As Object, e As EventArgs) Handles MyBase.Load
        For x As Integer = 1 To 5 'start five tasks
            Dim tsk As Task = Task.Run(Sub() ProcMyQ())
            MyTasks.Add(tsk)
        Next
    End Sub

    Private Sub ProcMyQ()
        Do While Not MyBC.IsCompleted
            Dim s As String = ""
            Try
                s = MyBC.Take 'block here
            Catch ex As InvalidOperationException
                ' means that Take() was called on a completed collection
                If MyBC.IsCompleted Then Exit Do
            End Try
            Debug.WriteLine(s)
        Loop
    End Sub

    Private Sub Button1_Click(sender As Object, e As EventArgs) Handles Button1.Click
        For x As Integer = 1 To 100
            MyBC.Add(x.ToString)
        Next
    End Sub

End Class

[nbrege]

Thanks, that seems to work. Just a quick question ... what's the reason for adding the tasks to a list?

[dbasnett]

Thanks, that seems to work. Just a quick question ... what's the reason for adding the tasks to a list?

Mostly habit. In case you need to know if all the tasks have completed. I should have shown that.

    Private Sub Form1_FormClosing(sender As Object, e As FormClosingEventArgs) Handles Me.FormClosing
        MyBC.CompleteAdding()
        Task.WaitAll(MyTasks.ToArray)
    End Sub

[nbrege]

Thanks again. I think I can make this work for my particular application. I never knew there was such a thing as a BlockingCollection until today.

[nbrege]

I have one more quick question ... is there a way to clear all the items from a BlockingCollection? I dont see a Clear method.

[dbasnett]

I have one more quick question ... is there a way to clear all the items from a BlockingCollection? I dont see a Clear method.

Thanks again. I think I can make this work for my particular application. I never knew there was such a thing as a BlockingCollection until today.

I don't see a Clear either. Here is an approach.

        While MyBC.TryTake(Nothing)
        End While

I think if .CompleteAdding is set Takes and Adss will fail.