2

Imagine there is a stream of requests for about 500 types of work. There can be say 5 workers in parallel. One type of work should be executed by at most one worker at the same time. The requests for a specific type of work should thus be done sequentially. So further requests for that active type of work should be enqueued till that current work is finished.

So, when the work of 1 type is being executed, only further requests for that type of work are enqueued. In the mean time all other types of work can continue.

In our Spring environment I tried a number of solutions for solving this situation like:

  • 2- Each type of work is assigned a queue. So there will be many queues. Sometimes a type of work will disappear, so then the specific queue is deleted. So any type of work is out on the specific queue.
  • 3- Have 1 message queue dispatching to a limited set of worker queues. That requires quite some logic for dispatching. There should be a standard solution for this problem.
  • 4- Have 1 message queue in combination with database tables that contain the 'suspended' messages. The problem of this solution is that it is hard to process the messages of one type of work in the right, sequential order.
  • 5- A bit awkward: just use a database table with workers reading the first (oldest) matching message. It works, but not ready with setting the right transaction boundaries. Is quite expensive in relation to database access

We started with experimenting with the dynamic creation of queues. This feels like very dynamically creating/suspending/resuming infrastructure. Is this right way to go?

The current solutions don't proof to be a good, solid solution.

How to model this in a good way? Is the option of dynamic creation, suspending and resuming of queues a good idea?

12
  • 1
    I'd consider your 4th solution first: having the workers maintain a shared list of currently-active types. However, there's a potential risk that this list becomes outdated in case of worker crashes or network interruptions.
    – amon
    Oct 26, 2022 at 21:13
  • Clarify "The requests for a specific type of work should thus be done sequentially. So further requests for an active type of work should be suspended till that work is finished." because as written you are saying that whatever is making the request cannot put the request on the queue, but the rest implies what you want is that the workers do not process any queued request, which are very different problems. Oct 27, 2022 at 11:08
  • When the work of 1 type is being executed, only further requests for that type of work are suspended. In the mean time all other types of work can continue.
    – tm1701
    Oct 27, 2022 at 11:12
  • 1
    Ok, so we have to assume that no request should be missed or lost.
    – Laiv
    Oct 28, 2022 at 7:30
  • 1
    Some folks in the mean time completed a POC with RabbitMq with option 2. And it works better than expected. My hesitation was that when the number of types of work grow, that this will be too heavy 'burden' for the infrastructure as a whole.
    – tm1701
    Nov 3, 2022 at 17:38

2 Answers 2

3
+50

We started with experimenting with the dynamic creation of queues. This feels like very dynamically creating/suspending/resuming infrastructure. Is this right way to go?

How to model this in a good way? Is the option of dynamic creation, suspending and resuming of queues a good idea?

It seems Netflix just built something very similar to your use case/problem: https://netflixtechblog.com/timestone-netflixs-high-throughput-low-latency-priority-queueing-system-with-built-in-support-1abf249ba95f

As I understood, they've built what they call an "Exclusive Queue", on which a "message" can be stored with its metadata. Within this metadata there is a "exclusivity key", which causes the message to be non-dequeuable if there is already a consumer processing a previous message with this same key and value.

For this solution, among all components, it comprises a Redis cluster (they represent and persist queues as Sorted Sets; they persist message payloads & Queue metadata in Hashes), Apache Kafka for publishing internal events of this solution and Elasticsearch to store primary and secondary indexes of messages (which mean "current" and "historical" messages, respectively).

Not sure if this helps, you can just downvote if it's not related.

2
  • Certainly worth having a look at! Oct 28, 2022 at 20:10
  • Some cool stuff as well. I have to study that more in depth.
    – tm1701
    Oct 28, 2022 at 20:13
2

I'd tackle the problem from different angles.

Is the problem really well-understood?

You didn't explain what the reason for wanting to handle tasks of one kind sequentially. Are there resource contention reasons, possible data corruption, etc.? Your question starts with what you think is the solution to the underlying problem, but you don't say what this problem really is.

Design an algorithm

An algorithm can be designed and analysed for theoretical correctness without committing to a specific technology. You will need some kind of atomic operations to ensure that no two workers work on the same kind of task.

You should also check whether your algorithm handles peripheral issues such as resource (worker) starvation, cost of switching between item types, timeliness etc.

Choose an implementation technology

Pure message brokers such as ActiveMQ or RabbitMQ probably won't handle your one-worker-per-type requirement well. A database typically would have to be polled, unless you can offload part of the work into triggers and stored procedures. I'd look at Redis for a nice mix of data structures and flexibility.

Rough implementation idea

Work items are queued into one queue per item type. Queues (lists) are cheap. In addition, the item type indicator is pushed into an additional queue from which idle workers fetch item types to work on.

Idle workers check this queue and assign themselves to an item type if no other worker handles it already. That's where you need atomic primitives to ensure that only one worker handles an item type.

Workers assigned to an item type pick work items from the respective queue and execute them. Once their queue is empty, they return into the idle state. To avoid resource starvation and ensure fairness, workers may stop handling a queue after a set time or number of items.

There are a number of possible race conditions which need to be considered, of course.

3
  • Thank you so far! Why the sequence is important? Any work per type of work should be done in the order the work request was created. Each request builds further on the results of the previous steps. No request can be left out. Per request of a type of work there will be an output file with a sequence number in it. Like, WorkType1_1.zip. WorkType1_2.zip. WorkType2_1.zip, etc.
    – tm1701
    Oct 28, 2022 at 10:05
  • When world you prefer JMS dynamic queues (with eg rabbitmq or active mq) over in memory concurrent queues? I would think of restartability, etc
    – tm1701
    Oct 28, 2022 at 10:40
  • Which implementation? We work in a Spring environment.
    – tm1701
    Oct 28, 2022 at 11:20

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.