How to implement a job queue with Redis
In how to cache with Redis, we implemented a simple cache backed by Redis.
That’s just one use case of Redis.
Redis is also used as a messaging server to implement the processing of background jobs or other kinds of messaging tasks.
This post explores implementing this pattern with Quarkus and the new Redis data source API.
Job Queues and Supes!
A job queue is a data structure storing execution requests. Job dispatchers submit the tasks they want to execute in that data structure. On the other side, job consumers poll the requests and execute them.
There are plenty of variants of that pattern, so let’s focus on the following application. We have an application managing heroes and villains. The application offers the possibility to simulate a fight between a random hero and a random villain. The fight simulation is delegated to fight simulators, applications dedicated to that task.
In this context, the main application submits the fight request to the job queue. Then, the fight simulators poll the submitted fight request and execute them.
The fight outcomes are communicated using another Redis feature: pub/sub communication. The simulators send the outcome to a channel consumed by the application. The application then broadcasts these outcomes to a web page.
This post only discusses the interaction with Redis. The rest of the application is straightforward and just uses RESTEasy Reactive and Hibernate ORM with Panache. You can find the full code of the application on https://github.com/cescoffier/quarkus-redis-job-queue-demo.
Submitting jobs
The first task is to model the job queue. We are using a Redis list to store the FightRequest.
package me.escoffier.quarkus.redis.fight;
public record FightRequest(String id, Hero hero, Villain villain) {
}Redis lists distinguish the left side of the list from the right side of the list. This distinction allows implementing a FIFO queue where we write on the left side and consume from the right side.
To manipulate a Redis list, we need the group of commands associated with this data structure.
In the SupesService class, we inject the RedisDataSource and retrieve the group of commands:
public SupesService(RedisDataSource dataSource, ...) {
commands = dataSource.list(FightRequest.class);
// ...
}Let’s now look at the submitAFight method:
public FightRequest submitAFight() {
var hero = Hero.getRandomHero();
var villain = Villain.getRandomVillain();
var id = UUID.randomUUID().toString();
var request = new FightRequest(id, hero, villain);
commands.lpush("fight-requests", request);
return request;
}The submitAFight method retrieves the random fighters, computes an id, builds the FightRequest instance, and executes the LPUSH command.
The LPUSH command writes the given item to the left side of the list stored at the given key (fight-requests).
Receiving the job requests
Let’s now look at the other side: the fight simulators.
The simulators poll the FightRequests from the Redis list representing our job queue and simulate the fight.
The simulator is implemented in me.escoffier.quarkus.redis.fight.FightSimulator.
The constructor receives a configured name (to distinguish multiple simulators) and the Redis data source.
It creates the objects to emit the Redis commands to read from a Redis list:
public FightSimulator(@ConfigProperty(name = "simulator-name") String name, RedisDataSource ds) {
this.name = name;
this.queue = ds.list(FightRequest.class);
// ...
}The simulator polls the fight requests and for each of them simulate the fight.
The implementation is an infinite loop (it only stops when the application is shut down).
In each iteration, it reads the pending FightRequest from the right side of the queue with the BRPOP command.
If there is no pending request, it restarts from the beginning of the loop.
If it has a request, it simulates the fight:
@Override
public void run() {
logger.infof("Simulator %s starting", name);
while ((!stopped)) {
KeyValue<String, FightRequest> item =
queue.brpop(Duration.ofSeconds(1), "fight-requests");
if (item != null) {
var request = item.value();
var result = simulate(request);
//...
}
}
}The BRPOP command retrieves and removes the last (right) element of the list.
Unlike the RPOP, it waits for a given amount of time (1 second in the code above) if there are no elements in the list.
So, if the list contains an element, it gets it.
Otherwise, it waits up to one second before giving up.
It returns null in this case.
The BRPOP command returns a KeyValue composed of the key of the list and the FightRequest.
It uses that structure because you can pass multiple keys, which is convenient when you have lists with priorities.
The BRPOP command also avoids spinning indefinitely if the list is empty, as it waits for 1 second during each iteration.
Finally, the BRPOP command is atomic.
It means that if you have multiple simulators, they cannot retrieve the same item.
It dispatches each item once.
Sending the fight outcome
The pool loop retrieves the FightRequests from the queue and simulates the fights, but how to communicate the results?
For this, we use another Redis feature: pub/sub communication.
In simple words, we are going to send the FightResult to a channel.
Applications subscribing to that channel will receive the emitted FightResult.
A FightResult contains the request id, the two fighters, and the name of the winner:
package me.escoffier.quarkus.redis.fight;
public record FightResult(String id, Hero hero, Villain villain, String winner) {
}To use Redis pub/sub commands, we need the object associated with this group.
In the FightSimulator, we also uses the pubsub method to get that object:
public FightSimulator(@ConfigProperty(name = "simulator-name") String name, Logger logger, RedisDataSource ds) {
this.name = name;
this.logger = logger;
this.queue = ds.list(FightRequest.class);
this.publisher = ds.pubsub(FightResult.class); // <--- this is it!
}Now, we can use this publisher to send the FightResults.
After each fight, we call publisher.publish to send the FightResult instance to the fight-results channel:
@Override
public void run() {
logger.infof("Simulator %s starting", name);
while ((!stopped)) {
KeyValue<String, FightRequest> item = queue.brpop(Duration.ofSeconds(1), "fight-requests");
if (item != null) {
var request = item.value();
var result = simulate(request);
publisher.publish("fight-results", result); // Send the outcome
}
}
}Receiving the fight outcome
At that point:
-
we submit the fight request into the job queue,
-
we consume that queue and simulate the fight,
-
we send the outcome to the
fight-resultschannel.
So, the only missing piece is the consumption of that channel.
Let’s return to the me.escoffier.quarkus.redis.supes.SupesService class.
In the constructor, we also inject the ReactiveRedisDataSource, the reactive variant of the Redis data source.
Then, in the constructor code, we subscribe to the fight-results.
public SupesService(RedisDataSource dataSource, ReactiveRedisDataSource reactiveRedisDataSource) {
commands = dataSource.list(FightRequest.class);
stream = reactiveRedisDataSource.pubsub(FightResult.class).subscribe("fight-results")
.broadcast().toAllSubscribers();
}Because we use the reactive data source, this subscription returns a Multi<FightResult>, ready to be served by Quarkus and an SSE (see SupesResource.java):
@GET
@Produces(MediaType.SERVER_SENT_EVENTS)
@RestStreamElementType(MediaType.APPLICATION_JSON)
public Multi<FightResult> fights() {
return supes.getFightResults();
}
.broadcast().toAllSubscribers() instructs Quarkus to broadcast all the received FightResult to all the connected SSE.
So, the browser filters out unrequested results.
|
Running the system
The circle is complete! The full code source is available from https://github.com/cescoffier/quarkus-redis-job-queue-demo. To run the system, open three terminals.
First, we start the supes-application.
In the first terminal, navigate to the supes-application and run mvn quarkus:dev
Quarkus automatically starts the PostgreSQL and Redis instances (if your machine can run containers).
In the console, hit h and then c.
It displays the running dev services.
Look for the redis one, and copy the quarkus.redis.hosts injected configuration:
redis-client - Up About a minute
Container: 348edec50f80/trusting_jennings docker.io/redis:7-alpine
Network: bridge - 0.0.0.0:53853->6379/tcp
Exec command: docker exec -it 348edec50f80 /bin/bash
Injected Config: quarkus.redis.hosts=redis://localhost:53853In the previous snippet, copy: quarkus.redis.hosts=redis://localhost:53853.
This is the address of the redis server.
We need to configure to the simulators with that address.
If you go to http://localhost:8080, the web page is served.
You can hit the fights! button a few times.
The fight won’t happen as we have no simulator. However, the fight requests have been submitted and stored in the list. So they are not lost.
Now, in the second terminal, navigate to the fight-simulator directory, and run:
mvn package
java -Dsimulator-name=A -Dquarkus.redis.hosts=redis://localhost:53853 -jar target/quarkus-app/quarkus-run.jarIMPORTANT: update the quarkus.redis-hosts with the one copied above.
As soon as you start it, it processes the pending fight requests:
2022-09-11 15:31:58,914 INFO [me.esc.qua.red.fig.FightSimulator] (Thread-3) Simulator A is going to simulate a fight between Pakku and Tulon Voidgazer
2022-09-11 15:31:59,786 INFO [me.esc.qua.red.fig.FightSimulator] (Thread-3) Simulator A is going to simulate a fight between Comet Zuko and Arishem The Judge (Knullified)
2022-09-11 15:32:01,809 INFO [me.esc.qua.red.fig.FightSimulator] (Thread-3) Simulator A is going to simulate a fight between Ms. America and Kazumi (Devil Form)If you go back to the web page, the winners get a halo:
Now, in the third terminal, navigate to the fight-simulator directory, and run:
java -Dsimulator-name=B -Dquarkus.redis.hosts=redis://localhost:53853 -jar target/quarkus-app/quarkus-run.jarIMPORTANT: as in the previous command, update the quarkus.redis-hosts with the one copied above.
Go back to the web page and click on the fight! button a few times.
Check the logs of both simulators to see that the fight requests are now dispatched beween the two simulators.
Summary
This posts explains how you can implement a job queue with Redis and the Quarkus Redis datasource API.
Learn more about the Redis data source API from the Quarkus documentation. We will publish more content about Redis patterns, so stay tuned!
