Using Sorted Sets Of Redis To Delay Execution In ASP.NET Core

Introduction

 

In a previous article, I showed you how to delay execution via keyspace notifications of Redis in ASP.NET Core, and I will introduce another solution based on Redis.

Sorted Sets, a data structure of Redis, also can help us work it out.

 

We can make a timestamp as score, and the data as value. Sorted Sets provides a command that can return all the elements in the sorted set  with a score between two special scores.

 

Setting 0 as the minimum score and current timestamp as the maximum score, we can get all the values whose timestamp are less than the current timestamp, and they should be executed at once and should be removed from Redis.

 

Taking a sample for more information.

 

Add some values at first.

ZADD task:delay 1583546835 "180"
ZADD task:delay 1583546864 "181"
ZADD task:delay 1583546924 "182"

Suppose the current timestamp is 1583546860, so we can get all values via the following command.

ZRANGEBYSCORE task:delay 0 1583546860 WITHSCORES LIMIT 0 1

We will get the value 180 from the above sample, and then we can do what we want to do.

 

Now, let's take a look at how to do this in ASP.NET Core.

 

Create Project

 

Create a new ASP.NET Core Web API project and install CSRedisCore.

<ItemGroup>
    <PackageReference Include="CSRedisCore" Version="3.4.1" />
</ItemGroup>

Add an interface named ITaskServices and a class named TaskServices.

public interface ITaskServices
{
    Task DoTaskAsync();

    Task SubscribeToDo();
}

public class TaskServices : ITaskServices
{
    public async Task DoTaskAsync()
    {
        // do something here
        // ...

        // this operation should be done after some min or sec

        var cacheKey = "task:delay";
        int sec = new Random().Next(1, 5);
        var time = DateTimeOffset.Now.AddSeconds(sec).ToUnixTimeSeconds();
        var taskId = new Random().Next(1, 10000);
        await RedisHelper.ZAddAsync(cacheKey, (time, taskId));
        Console.WriteLine($"{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")} done {taskId} here - {sec}");
    }

    public async Task SubscribeToDo()
    {
        var cacheKey = "task:delay";
        while (true)
        {
            var vals = RedisHelper.ZRangeByScore(cacheKey, -1, DateTimeOffset.Now.ToUnixTimeSeconds(), 1, 0);

            if (vals != null && vals.Length > 0)
            {
                var val = vals[0];

                // add a lock here may be more better
                var rmCount = RedisHelper.ZRem(cacheKey, vals);

                if (rmCount > 0)
                {
                    Console.WriteLine($"{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")} begin to do task {val}");
                }
            }
            else
            {
                await Task.Delay(500);
            }
        }
    }
}

Here we will use DateTimeOffset.Now.AddSeconds(sec).ToUnixTimeSeconds() to generate the timestamp, the sec parameter means that we should execute the task after some seconds.

 

For the delay execution, it will poll the values from Redis to consume the tasks, and make it sleep 500 milliseconds if cannot get some values.

 

When we get a value from Redis, before we execute the delay task, we should remove it from Redis at first.

 

Here is the entry of this operation.

[ApiController]
[Route("api/tasks")]
public class TaskController : ControllerBase
{
    private readonly ITaskServices _svc;

    public TaskController(ITaskServices svc)
    {
        _svc = svc;
    }

    [HttpGet]
    public async Task<string> Get()
    {
        await _svc.DoTaskAsync();
        return "done";
    }
}

We will put the subscribe to a BackgroundService

public class SubscribeTaskBgTask : BackgroundService
{
    private readonly ILogger _logger;
    private readonly ITaskServices _taskServices;

    public SubscribeTaskBgTask(ILoggerFactory loggerFactory, ITaskServices taskServices)
    {
        this._logger = loggerFactory.CreateLogger<RefreshCachingBgTask>();
        this._taskServices = taskServices;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        stoppingToken.ThrowIfCancellationRequested();
        await _taskServices.SubscribeToDo();
    }
}

At last, we should register the above services in startup class.

public class Startup
{
    // ...

    public void ConfigureServices(IServiceCollection services)
    {
        var csredis = new CSRedis.CSRedisClient("127.0.0.1:6379");
        RedisHelper.Initialization(csredis);

        services.AddSingleton<ITaskServices, TaskServices>();
        services.AddHostedService<SubscribeTaskBgTask>();

        services.AddControllers();
    }
}

Here is the result after running this application.

 

 

Here is the source code you can find in my GitHub page.

• DelayDemo2

Summary

 

This article showed you a simple solution of how to delay execution in ASP.NET Core using Redis sorted sets.

I hope this will help you!