RabbitMQ dequeue multiple messages at time

Question

I working with RabbitMQ in .net core (latest version from NuGet). I have a queue, with priority.

My code is inserting 10 messages and dequeues 10 messages. To dequeue the messages, I used EventingBasicConsumer which gets the messages on Push.

This is my code:

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;

namespace Test
{
    class Program
    {
        public static IConnection Connection = GetConnection("localhost", "xxx", "xxx");
        public static IModel Channel = Connection.CreateModel();
        public static void Main(string[] args)
        {
            IDictionary<String, Object> args2 = new Dictionary<String, Object>();
            args2.Add("x-max-priority", 256);
            Channel.QueueDeclare("IDG", true, false, false, args2);

            for (int i = 0; i < 10; i++)
            {
                Send("IDG", (i % 10).ToString(), (byte)(i % 10));
                Console.WriteLine("Queued: "+ (i % 10).ToString());
            }

            Receive("IDG", Channel);
            Console.ReadLine();
        }

        public static IConnection GetConnection(string hostName, string userName, string password)
        {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.HostName = hostName;
            connectionFactory.UserName = userName;
            connectionFactory.Password = password;
            return connectionFactory.CreateConnection();
        }

        public static void Send(string queue, string data, byte priority)
        {
            var properties = Channel.CreateBasicProperties();
            properties.Priority = priority;
            Channel.BasicPublish(string.Empty, queue, properties, Encoding.UTF8.GetBytes(data));
        }

        public static void Receive(string queue, IModel channel)
        {
            IDictionary<String, Object> args2 = new Dictionary<String, Object>();
            args2.Add("x-max-priority", 256);
            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += Consumer1_Received;
            channel.BasicConsume(consumer, queue, autoAck: false, arguments: args2);
        }

        private static void Consumer1_Received(object sender, BasicDeliverEventArgs e)
        {
            var message = Encoding.UTF8.GetString(e.Body);
            Console.WriteLine($"{DateTime.Now}: [x] Received {message}");

            // Simulate Processing...
            Thread.Sleep(3000);

            Channel.BasicAck(e.DeliveryTag, false);
        }
    }
}

The output:

Queued: 0
Queued: 1
Queued: 2
Queued: 3
Queued: 4
Queued: 5
Queued: 6
Queued: 7
Queued: 8
Queued: 9
16-Jul-19 11:37:59 AM: [x] Received 9
16-Jul-19 11:38:02 AM: [x] Received 8
16-Jul-19 11:38:05 AM: [x] Received 7
16-Jul-19 11:38:08 AM: [x] Received 6
16-Jul-19 11:38:11 AM: [x] Received 5
16-Jul-19 11:38:14 AM: [x] Received 4
16-Jul-19 11:38:23 AM: [x] Received 3
16-Jul-19 11:38:23 AM: [x] Received 2
16-Jul-19 11:38:23 AM: [x] Received 1
16-Jul-19 11:38:23 AM: [x] Received 0

Processing of each messages takes 3 seconds (see Thread.Sleep(3000) in Consumer1_Received delegate).

The question

I want to process multiple messages on same time. Let's say that my machine can handle 5 messages simulatnicly. So, processing the whole 10 messages should takes approximatly 6 seconds.

1. How to do define the concorency amount of running messages at same time?
2. How to pull N messages? I saw the method BasicGet() which pulling a single message. Is this possible to pull more then a message?

UPDATE 1

I already tried to use multiple consumer, but the throuput was the same. This is the code:

public static void Receive(string queue, IModel channel)
{
    IDictionary<String, Object> args2 = new Dictionary<String, Object>();
    args2.Add("x-max-priority", 256);

    channel.BasicQos(0, 1, true);

    var consumer1 = new EventingBasicConsumer(channel);
    consumer1.Received += Consumer_Received;
    channel.BasicConsume(consumer1, queue, autoAck: false, arguments: args2);

    var consumer2 = new EventingBasicConsumer(channel);
    consumer2.Received += Consumer_Received;
    channel.BasicConsume(consumer2, queue, autoAck: false, arguments: args2);
}

private static void Consumer_Received(object sender, BasicDeliverEventArgs e)
{
    var message = Encoding.UTF8.GetString(e.Body);
    Console.WriteLine($"{DateTime.Now}: [x] Received {message}");

    // Simulate Processing...
    Thread.Sleep(3000);

    Channel.BasicAck(e.DeliveryTag, false);
}

And the output is: (still 3 seconds\message)

16-Jul-19 12:29:30 PM: [x] Received 6
16-Jul-19 12:29:33 PM: [x] Received 5
16-Jul-19 12:29:36 PM: [x] Received 4
16-Jul-19 12:29:39 PM: [x] Received 3
16-Jul-19 12:29:45 PM: [x] Received 2
16-Jul-19 12:29:51 PM: [x] Received 1
16-Jul-19 12:30:00 PM: [x] Received 0

Answer 1

You use the same IModel instance for both consuming and producing. Create two separate models. Thread.Sleep(3000) blocks dedicated rabbitmq thread used to receive new messages.