Build an Async Queue System using Redis & ASP.NET

The perfect project for Redis beginners

If you’re new to Redis and have just enough experience in ASP.NET, then this is an amazing project to familiarize yourself with the concepts behind Redis.

We will be building an Async Queue System throughout this article. This system will support adding items to a queue and processing them later on.

It’s Async because processing items in the queue is not coupled with putting items in the queue. You can put items in the queue and not have to worry about when they are going to be processed. This is also commonly referenced as “Fire and Forget.”

Setting Up the Project

To make things more interactive, the project will be built as a Web API.

By using the .NET CLI, you can quickly scaffold one:

dotnet new webapi -o QueueAPI

We only require a single external library, which is Redis.

dotnet add package StackExchange.Redis

All jobs inside the queue will have the same properties, which are encapsulated into the following class:

public class JobModel
{
    public string Id { get; set; } = Guid.NewGuid().ToString();
    public string Name { get; set; } = string.Empty;
    public string Status { get; set; } = string.Empty;
}

Each job has a unique ID, name, and status - “queued” or “in progress”.

Registering Redis into ASP.NET

In order to use Redis with any service in the application at a later stage, it is necessary to register it as a Singleton.

You will also need to provide the login credentials to the Redis instance, whether it is hosted on a cloud or locally.

// Program.cs
var connectionOptions = new ConfigurationOptions
{
    EndPoints = { "<your-redis-endpoint>" },
    Password = "<your-redis-password>"
};

builder.Services.AddSingleton<IConnectionMultiplexer>(ConnectionMultiplexer.Connect(connectionOptions));

In practice, the connection between Redis and the API we are building will be handled by a “Connection Multiplexer,” which we’ll use to obtain an instance of the Redis Database later on.

In case you don’t have a Redis Database, you can get one for free hosted in the cloud at https://app.redislabs.com/.

In the Redis Queue we’re going to build, all new jobs will be added to the tail of the queue, and jobs will be processed from the head of the queue.

Adding Jobs to the Queue

We have everything ready to start interacting with Redis. The next step is to create a class and inject Redis into it so that we can add things to the queue.

// JobEnqueuerService.cs

public class JobEnqueuerService : IJobEnqueuerService
{
    private readonly IDatabase _redisDatabase;
    public JobEnqueuerService(IConnectionMultiplexer multiplexer)
    {
        _redisDatabase = multiplexer.GetDatabase();
    }
}

The service will have two methods: one to add a job to the queue and the other to fetch all jobs currently in the queue, along with their status.

// JobEnqueuerService.cs

// Add job at the back of the queue
public async Task EnqueueJobAsync(JobModel job)
{
    await _redisDatabase.ListLeftPushAsync("jobQueue", JsonSerializer.Serialize(job));
    await _redisDatabase.HashSetAsync(job.Id, "status", "queued");
}

// Fetch all jobs in the queue, along with their status
public async Task<List<JobModel>> GetJobsAsync()
{
    var jobs = await _redisDatabase.ListRangeAsync("jobQueue");
    var jobList = new List<JobModel>();
    foreach (var job in jobs)
    {
        var redisJob = JsonSerializer.Deserialize<JobModel>(job);
        redisJob.Status = _redisDatabase.HashGet(redisJob.Id, "status");
        jobList.Add(redisJob);
    }
    return jobList;
}

Building the API Endpoints

To call the two methods we just built, we’ll build two simple API endpoints that can be called using Swagger, Postman, or Insomnia.

// JobEnqueuerController.cs

public async Task<IActionResult> EnqueueTask(JobModel job)
{
    await _jobEnqueuer.EnqueueJobAsync(job);
    return Ok();
}

[HttpGet]
public async Task<IActionResult> GetTasks()
{
    var tasks = await _jobEnqueuer.GetJobsAsync();
    return Ok(tasks);
}

Now that we have the two methods in a separate class, we can go back to the controller and update the two API endpoints.

// JobEnqueuerController.cs

[ApiController]
[Route("[controller]")]
public class JobEnqueuerController : ControllerBase
{
    // Inject the service
    private readonly IJobEnqueuerService _jobEnqueuer;

    public JobEnqueuerController(IJobEnqueuerService jobEnqueuer)
    {
        _jobEnqueuer = jobEnqueuer;
    }

    [HttpPost]
    public async Task<IActionResult> EnqueueTask(JobModel job)
    {
        await _jobEnqueuer.EnqueueJobAsync(job);
        return Ok();
    }
    
    [HttpGet]
    public async Task<IActionResult> GetTasks()
    {
        var tasks = await _jobEnqueuer.GetJobsAsync();
        return Ok(tasks);
    }
}

Make sure to register the service as scoped before running the application:

// Program.cs

builder.Services.AddScoped<IJobEnqueuerService, JobEnqueuerService>();

Processing Jobs from the Queue

Now that we have jobs in the queue, we can process them one by one and remove them!

This will be done by a BackgroundService in ASP.NET, which runs constantly in the background. The service will constantly check if there are new jobs in the queue, and process the oldest one.

This class has two methods: one to fetch the oldest job in the queue and another to remove it.

// JobDequerService.cs

public async Task<JobModel?> DequeueJobAsync()
{
    var job = await _redisDatabase.ListGetByIndexAsync("jobQueue", 0);
    if (!job.HasValue)
    {
        return null;
    }
    var redisJob = JsonSerializer.Deserialize<JobModel>(job);
     
    // change status from "queued" to "in progress"
    await _redisDatabase.HashSetAsync(redisJob.Id, "status", "in progress");
    return redisJob;
}

public async Task CompleteJobAsync(JobModel job)
{
    await _redisDatabase.ListRemoveAsync("jobQueue", JsonSerializer.Serialize(job));
    await _redisDatabase.KeyDeleteAsync(job.Id);
}

By inheriting from BackgroundService, you can define what logic needs to be constantly run in the background in a method named ExecuteAsync:

// JobDequerService.cs

// Inject Redis Database
private readonly IDatabase _redisDatabase;

public JobDequerService(IConnectionMultiplexer multiplexer)
{
    _redisDatabase = multiplexer.GetDatabase();
}

protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
    while (!stoppingToken.IsCancellationRequested)
    {
        var job = await DequeueJobAsync();
        if (job != null)
        {
            // Simulate time to complete job
            await Task.Delay(5000, stoppingToken);
            await CompleteJobAsync(job);
        }
    }
}

Make sure that the service is registered as a Hosted Service:

// Program.cs

builder.Services.AddHostedService<TaskDequerService>();

And that’s it! You can now play with it via Swagger or by building a user interface in your favorite front-end framework. You can also use the service Redis Insights to view the queue entries via a pretty interface.

Thank you so much for reading this article about Redis! 💻❤️

If you’d like to be up to date and read more articles like this one, be sure to follow me! In the meantime, you can read more interesting articles on my Medium profile 😀

See you in the next one! 👋