NServiceBus.RabbitMQ
NServiceBus.RabbitMQ copied to clipboard
Particular
Priority Queue Support
NServiceBus core does not support message priority. However, RabbitMQ does support priority queues.
In order to make this work the following needs to be implemented:
- Ability to create a queue as a priority queue (routing topology)
- Ability to assign a message priority based on a message type and / or value
This request is from a customer having integrations with third parties where the third party API's are used for multiple message types. These are often rate limited and they want to prioritize which message are delivered first.
Reference:
- https://www.rabbitmq.com/priority.html
- https://groups.google.com/forum/#!msg/particularsoftware/-Vg2EgCCmNk/hOigCnM9KgAJ
Interesting. Do we know if/how this is exposed in the RabbitMQ client nupkg?
It seems this is exposed via the current client. We could perhaps add support for it in DirectRoutingTopology.
I like the idea of supporting it, however I wonder how the API for setting a priority per message would look like? Would it be per message (while sending), per message type (in some configuration)? How it would be set? Using headers?
We could support it in both Topologies, as it is indifferent of topology
There are several things that would need to be addressed to be able to support this.
The queues themselves need to be declared with the x-max-priority argument, which includes passing in a value to set the number of priority levels. If we wanted our topologies to have knowledge of this value, then we'd have to have some way to specify the number of priority values per-queue.
According to the docs page, there is some overhead in setting this up, so I don't know that we'd want to just enable it for all queues, so that would mean some way to specify enabling for specific queues as well.
We also couldn't just have the topologies start declaring this value for all queues (with a predetermined number of priority levels) because that would be a breaking change to the topology since our QueueDeclare calls would not match the already existing queues.
We could not modify our queue creation at all and let end users manually set up the queues with the desired priority values. It would then be up to them to not run the installers to declare the queues, since they would fail at that point.
As @WojcikMike mentioned, we would then need some way to specify the message priority on a per-message basis as well, and then translate that to setting the BasicProperties.Priority property in BasicPropertiesExtensions.Fill.
The other interesting aspect of supporting this is this section from their docs page:
So if such a hungry consumer connects to an empty queue to which messages are subsequently published, the messages may not spend any time at all waiting in the queue. In this case the priority queue will not get any opportunity to prioritise them.
So if endpoints are consuming messages as soon as they come in, they won't actually see a benefit from supporting this!
So basically to support it in existing Topologies we would have to recreat queues as according to docs:
Queues can never change the number of priorities they support.
I hoped that we could update it with some script. That would be problematic.
Regarding:
So if endpoints are consuming messages as soon as they come in, they won't actually see a benefit from supporting this!
Yes. If you deal with the messages as they come the priority is useless. It only make any sense if you get big load of messages and there is a time to prioritize them so that consumer pick message based on priority.
I'd like to validate Priority Queue Support as the solution for the customer scenario mentioned.
@Particular/rabbitmq-transport-maintainers I propose we have a quick call with @ramonsmits about this next week so he can explain more about the specific customer scenario.
Is there any word on this?
I have Sagas where we're trying to solve the supervisor scenario (see links below) by sending thousands of command messages from the Saga to another queue. However, this blocks single messages on that same endpoint from getting processed until the bulk messages are processed. It would be very helpful if we could mark those 1000 messages from the saga as low priority so the system remains responsive for individual commands.
Secondly, if the Saga and the individual message handler are both running on the same endpoint, then timeouts won't work as the timeout message is sitting behind 1000s of messages. I want a timeout that runs every 5 seconds so I can check the status of the Saga operation, but I can't do that if timeouts aren't prioritized above the 1000 messages.
https://docs.particular.net/nservicebus/sagas/concurrency https://lostechies.com/jimmybogard/2014/02/27/reducing-nservicebus-saga-load/
Hi @NArnott, sorry for the delay. In your scenario we usually recommend splitting the handlers/sagas across endpoints appropriately. E.g. if you have contention between a saga and a stateless handler, you could split those into two separate endpoints so that they each have their own queue. We'd be happy to discuss your case in more detail. Just raise a support request and we'll get back to you.
Hi, guys. Is there are any plans to implement this feature in future releases?
Hi @mixtura to be honest, we are on the edge with this one. Till now we are believing that using priority queues bring very little value, and we have troubles seeing in what business cases it is a better solution than what @adamralph proposed in the comment above
For now, we keep this issue open but we are not planning to implement it in near future. If we find a case that would change our perspective on that topic we will revisit this approach.
Our usecase for this is that we have commands managing our batch processing. Sometimes we have commands that comes in that requires several hours of intensive processing to handle, effectively blocking the rest of the queue from doing anything meanwhile. Meanwhile we also have other much more lightweight tasks coming in, that only takes in the order of seconds to process, so being able to prioritize these first for a more responsive system would be a lot of value for us.
We cannot determine the routing infrastructure ahead of the time, since all the batch job descriptions is just passed through one big "ExecuteJob" command. It would be a horrible pain to have to manage a dynamic number of priorities using different queues in this case, rather than just being able to attach a priority to the command, and have Rabbit deliver things in the proper order.
@zlepper It's not clear to me how priority queues would help in your scenario. You'd have to be able to evaluate ahead of time the size of the job in order to assign it a priority. If you're able to do that, then you could use that same logic to send the message to a "high priority" or "low priority" endpoint for processing.
Alternatively, you could consider adding additional instances of the processing endpoint, allowing more messages to be processed that way.
All of which are things that rabbit already has the logic for handling internally, without us having to jump through hoops to reimplement.
Also if it's separate full queues, then the normal message rate limiting would not apply I assume? So I might end up getting messages from both the high and low priority at the same time?
And scaling up instances is sadly not an option for us, as the software is run on premise by most of our customers.
@zlepper Here is why we question the usefulness of priority queues:
If you have one of these messages that creates a blocking job, it may be possible it will never be worked on if faster higher priority messages are always available. This is the opposite of the scenario you describe, but I assume this isn't a desirable state, assuming it is applicable to your system.
If it isn't, the other scenario we can see happening is where those higher priority messages are eventually processed and now the blocking blocking job messages are started. You would still end up with a scenario where all your available "processing slots" are filled with these blocking jobs that have been waiting, and now any newer higher priority jobs (messages) would be blocked. So you are back to being unresponsive.
What is left over are scenarios where what is sent to the queue is orderly enough to always "work out" with priority queues. We have yet to come across an actual customer use case that wouldn't end up in one of the 2 previous scenarios if we implemented priority queues. If your scenario is that orderly, it may be that a properly written Saga could achieve the same result with NServiceBus.
Ultimately, what most customers that ask for priority queues want is for priority work to continue, but limit how many blocking jobs you are running at any given time so that the higher priority work is always responsive.
In this case, with NServiceBus, I would recommend (as others have done) separate endpoints with a dynamic routing policy based on the nature of the job.
It is tricky to do, but since you have limitations with your customers on-premise deployments, you can host multiple endpoints in one process. You could then tune the message processing appropriately on each to make sure the higher priority jobs are responsive, but that long running jobs have some capacity to do work.
Ideally you would route based on message types, however an alternative would be to override the default routing during Send. This would be based on the same logic you would now use to assign a priority, but just set the endpoint name override for the message send.
Perhaps I missed the mark. It may best to set up a meeting with us so we could talk with you specifics about your system / requirements. If that is the case I would suggest contacting via email at [email protected].
If it isn't, the other scenario we can see happening is where those higher priority messages are eventually processed and now the blocking blocking job messages are started. You would still end up with a scenario where all your available "processing slots" are filled with these blocking jobs that have been waiting, and now any newer higher priority jobs (messages) would be blocked. So you are back to being unresponsive.
According to the requirements I have gotten, this is exactly the behavior they want, and are aware of the consequences of said behavior. As long as once the blocking job finished, the high priority jobs will be taken before low priority again, then everything is exactly how we have speced things out.
We cannot determine the routing infrastructure ahead of the time, since all the batch job descriptions is just passed through one big "ExecuteJob" command. It would be a horrible pain to have to manage a dynamic number of priorities using different queues in this case, rather than just being able to attach a priority to the command, and have Rabbit deliver things in the proper order.
What if you send the command to two endpoints. One for high-intensity work, the other for short work. It's even simpler when you publish events instead of actually sending commands to different endpoints. You'd be able in the future, to split up the messages between high, medium and low-intensity work or even further. All that without modifying the sending endpoint.
The idea is that each endpoint would figure out on its own, if it needs to process messages. The low-intensity endpoint, which receives lots of work but is done with it quickly, can dismiss all the messages containing high-intensity work. The other one can dismiss the other messages.
The idea is that the sender doesn't need to know where to actually send each message to.
Another idea is that you put a man-in-the-middle. For example if you need to query the database to be able to figure out which ones are highly-intensive and which are not. Have that endpoint send messages to an appropriate endpoint.
According to the requirements I have gotten, this is exactly the behavior they want, and are aware of the consequences of said behavior. As long as once the blocking job finished, the high priority jobs will be taken before low priority again, then everything is exactly how we have speced things out.
What if you could provide both? Have both high priority and low priority messages be continuously processed, without them blocking each other? The above scenarios or a variation on it, might offer exactly that.
I created a demo some time ago, which shows high-priority and low-priority queues using NServiceBus events. You can find it here: https://github.com/Particular/Workshop/tree/demos/demos/06-scaling-and-priority-queues
I have a business scenario where priority messages are a requirement.
- We have a third party product that allows us 10 concurrent licenses
- We are using RabbitMQ and competing consumers. (1 generic message Q that everyone sends to, and 2 consumer q's set to a throughput of 5 that are each tied to a specific license server).
- We have 100's of tasks that are kicked off in the morning and are pushed to generic Q.
- Sometimes a user wants to run a one off scenario manually.
The choice is to either segregate the licenses by adding a 3rd special 'priority' Q and reserve one or two licenses for end users, and thus not be able to use them for the 100's that run every morning, or have the users requests wait at the back of the Q before they get pulled off; potentially having to wait an hour or more.
If priority messaging was available, tasks initiated by a user could jump to the front of the Q; allowing us to maximize license usage and still allow users to not have to wait for all of the morning tasks to finish.
@macdonald-keith-vmware If the license is deployment-bound then you host both a high-priority endpoint and low-priority endpoints in the same process as we show in the endpoint multi-hosting sample.
@macdonald-keith-vmware If the license is deployment-bound then you host both a high-priority endpoint and low-priority endpoints in the same process as we show in the endpoint multi-hosting sample.
That doesn't 'really' solve the issue. There are a fixed 10 licenses that can be used. If the low-priority queue is using 10, and you send a message to the high-priority queue, it will exceed the license count, fail, and be re-queued to be retried. Imagine there are 100 messages in the 'low' priority queue, and 1 in the 'high', they are both using the same limited set of resources. Unless you reserve a specific license for the 'high' queue, you don't solve the issue. And if you do reserve specific licenses for that queue, then the 'low' priority queue has less it can work with when no-one needs the high priority queue
So in this case it is not deployment bound, but concurrency bound, such as a 3rd party remote integration that only allows 10 connections at a time or something similar. I suspected that may be the case, and while we still consider splitting endpoints as the ideal solution in most cases, we can see how this type of concurrency binding resource could cause problems with that approach.
I have already updated our internal issue for tracking the Priority Queues concept citing your request. This helps us prioritize working on new features / products. Unfortunately I do not have any ETA to share with you or the others in this thread on when we would begin working on this. We will update this issue when we have a status change we can share.
Priority queues are not supported by quorum queues. Since quorum queues are the default queues, and classic queues should no longer be used. This issue can be closed.
