I have ConcurrentQueue of type IProducerConsumerCollection i.e.
IProducerConsumerCollection<Job> _queue = new ConcurrentQueue<Job>();
and producer method which adds jobs to _queue and consumer method which processes the Job from _queue. Now in the consumer method, I like to process the jobs concurrently. Below is code for sample class with producer and consumer methods:
public class TestQueue
{
IProducerConsumerCollection<Job> _queue = new ConcurrentQueue<Job>();
private static HttpClient _client = new HttpClient();
public TestQueue()
{
WorkProducerThread();
WorkConsumerThread();
}
public void WorkConsumerThread()
{
if (_queue.Count > 0)
{
//At this point, 4 partitions are created but all records are in 1st partition only; 2,3,4 partition are empty
var partitioner = Partitioner.Create(_queue).GetPartitions(4);
Task t = Task.WhenAll(
from partition in partitioner
select Task.Run(async () =>
{
using (partition)
{
while (partition.MoveNext())
await CreateJobs(partition.Current);
}
}));
t.Wait();
//At this point, queue count is still 20, how to remove item from _queue collection when processed?
}
}
private async Task CreateJobs(Job job)
{
HttpContent bodyContent = null;
await _client.PostAsync("job", bodyContent);
}
public void WorkProducerThread()
{
if (_queue.Count == 0)
{
try
{
for (int i = 0; i < 20; i++)
{
Job job = new Job { Id = i, JobName = "j" + i.ToString(), JobCreated = DateTime.Now };
_queue.TryAdd(job);
}
}
catch (Exception ex)
{
//_Log.Error("Exception while adding jobs to collection", ex);
}
}
}
}
public class Job
{
public int Id { get; set; }
public string JobName { get; set; }
public DateTime JobCreated { get; set; }
}
There are 2 problems,
Partitioner.Create(_queue).GetPartitions(4); Partitioner.GetPartionscreates 4 partitions but all records are in 1st partition only; 2,3,4 partition are empty. I can't find, why this is happening? Ideally, all 4 partitions should have 5 records each (since total 20 records are in queue). I read this article from MSDN on partitioning but didn't get any clue. Also I checked the partitioning example from this article.Also, I want to remove the item from _queue after processing in consumer method and there is only one way _queue.TryTake method to remove item. I don't know how to remove item along with partitioning?
I can consider any alternate way to achieve the same result.
Thanks in advance.