RabbitMQ and C #

Question

RabbitMQ has a way to use it like MSSMQ, where you can pull 1000 messages from the queue, and then insert your inserts into the database and continue from there.

I can’t do this with a channel subscription, and then do foreach through BasicDeliveryEventArgs in the subscription, with what the If statement does with the maximum number of messages that I want to process at the moment.

Thanks in advance. However, it still accepts all 22k messages from the queue.

using (IConnection connection = factory.CreateConnection()) { using (IModel channel = connection.CreateModel()) { channel.QueueDeclare("****", true, false, false, null); var subscription = new Subscription(channel, "****", false); int maxMessages = 5; int i = 0; foreach (BasicDeliverEventArgs eventArgs in subscription) { if (++i == maxMessages) { Console.WriteLine("Took 5 messages"); subscription.Ack(eventArgs); break; } } } } 

Answer 1

I assume that you want to optimize the loading of messages into the database by grouping groups of them into larger transactions, and not into the transaction cost per message. With a mandatory warning that this means that large groups of messages can work together, even if only one of them causes a problem, here's how you do it ...

Install QOS on the channel:

 channel.BasicQos(0, 1000, false); 

This will allow you to pre-receive 1000 messages and block further traffic until you activate something. Note that it is not retrieved in blocks of 1000. Rather, it ensures that a maximum of 1000 UNACK messages are preloaded at any given time. Simulating block transfers is as simple as processing 1000 messages first and then ACK'ing them all at once.

See here and here for a more authoritative explanation than mine.

Another point. You might want to drop the queue as soon as messages are available, even if you have not made your quota of 1000 messages. You must do this by calling queue.BasicGet() inside the foreach until it ends, and then deliver everything you have (including the message you pulled from the subscription ) to the database. Caution: I have not tried this myself, so I could say garbage, but I think it will work. The beauty of this method is that it instantly sends messages to the database, without waiting for a full batch of 1000 messages. If the database lags behind processing too many small transactions, the lag from the prefetch will simply fill more between each cycle.