public DateTime CreateDate { get; set; }

public string Category { get; set; }

public object Payload { get; set; }

- Step 3 - Create ASP.NET Core Web API project and name it `JobQueue.ProducerService`
- Step 4 - Reference shared class library project to producer service project

- Step 5 - Create message producer

Let's create a message producer class that generates a random message. I used an excellent library [Bogus](https://github.com/bchavez/Bogus) to generate fake data for testing purposes.
```csharp


internal class MessageGenerator
{
    private static readonly string[] Categories = { "express", "normal" };
    private static readonly Faker<MessageModel> Faker;

    static MessageGenerator()
    {
        var random = new Random();

        Faker = new Faker<MessageModel>()
            .StrictMode(false)
            .RuleFor(p => p.Category, f => f.PickRandom(Categories))
            .RuleFor(p => p.MessageId, f => f.Random.Guid())
            .RuleFor(p => p.CreateDate, f => f.Date.Between(DateTime.Now.AddSeconds(-random.Next(1, 5)), DateTime.Now));
    }

    public static IEnumerable<MessageModel> GenerateMessages()
    {
        return Faker.Generate(100);
    }
}

internal class MessageStore
{
    private readonly List<MessageModel> _store = new();
    private static readonly MessageStore _instance = new();

    private MessageStore()
    {
    }

    public static MessageStore Instance => _instance;

    public int Count => _store.Count;

    public void AddMessages(IEnumerable<MessageModel> messages)
    {
        _store.AddRange(messages);
    }

    public IEnumerable<MessageModel> GetMessages(int count)
    {
        var message = _store.Take(count).ToList();
        _store.RemoveRange(0, message.Count);

        return message;
    }
}

internal class MessageProducerHostedService : IHostedService, IDisposable
{
    private Timer _timer;

    public Task StartAsync(CancellationToken cancellationToken)
    {
        _timer = new Timer(SeedData, null, TimeSpan.Zero, TimeSpan.FromSeconds(1));

        return Task.CompletedTask;
    }

    public Task StopAsync(CancellationToken cancellationToken)
    {
        _timer?.Change(Timeout.Infinite, 0);

        return Task.CompletedTask;
    }

    public void Dispose()
    {
        _timer?.Dispose();
    }

    private void SeedData(object state)
    {
        if (MessageStore.Instance.Count > 2000)
            return;

        var messages = MessageGenerator.GenerateMessages();
        MessageStore.Instance.AddMessages(messages);
    }
}

####Consumer Service
- Step 7 - Create ASP.NET Core Web API project and name it `JobQueue.ConsumerService`

- Step 8 - Install `Hangfire.AspNetCore` and `HangFire.Redis.StackExchange` nuget packages

- Step 9 - Implement a background service that periodically fetches messages from the producer service and enqueue into a queue
```csharp


public class MessageReceiverHostedService : IHostedService
{
    private readonly CancellationTokenSource _cts;
    private readonly IServiceProvider _serviceProvider;

    public MessageReceiverHostedService(IServiceProvider serviceProvider, ILogger<MessageReceiverHostedService> logger)
    {
        _serviceProvider = serviceProvider;
        _cts = new CancellationTokenSource();
    }

    public async Task StartAsync(CancellationToken cancellationToken)
    {
        await Task.Factory.StartNew(() => FetchMessagesAsync(_cts.Token), cancellationToken);
    }

    public Task StopAsync(CancellationToken cancellationToken)
    {
        _cts.Cancel();

        return Task.CompletedTask;
    }

    private async Task FetchMessagesAsync(CancellationToken cancellationToken)
    {
        while (true)
        {
            using var scope = _serviceProvider.CreateScope();
            var httpClient = scope.ServiceProvider.GetRequiredService<JobHttpClient>();
            var messages = await httpClient.GetJobMessagesAsync(cancellationToken);

            if (!messages.Any())
                continue;

            var categories = messages.GroupBy(m => m.Category).ToList();

            Parallel.ForEach(categories, category =>
            {
                Enqueue(category.Key, category.ToList());
            });

            await Task.Delay(TimeSpan.FromSeconds(5), cancellationToken);

            if (cancellationToken.IsCancellationRequested)
                break;
        }
    }

    private void Enqueue(string queueName, List<MessageModel> messages)
    {
        var client = new BackgroundJobClient();
        var state = new EnqueuedState(queueName);

        foreach (var message in messages.OrderBy(o => o.CreateDate))
        {
            Expression<Action> action = queueName == "express"
                ? () => MessageProcessor.ProcessExpressMessageAsync(message, message.MessageId)
                : () => MessageProcessor.ProcessNormalMessageAsync(message, message.MessageId);
            client.Create(action, state);
        }
    }
}

[Queue("normal")]
[DisplayName("JobId: {1}")]
[AutomaticRetry(Attempts = 3)]
public static async Task ProcessNormalMessageAsync(MessageModel message, Guid messageId)
{
    await Task.Delay(TimeSpan.FromSeconds(new Random().Next(1, 4)));
}

We have to methods for processing messages from each queue. One limitation of Hangfire is that you cannot use a method to process jobs from multiple queues (or at least I'm not aware of it but it can be done by implementing a custom `Queue` attribute).

- Step 9 - Reference shared class library project to consumer service project

- Step 10 - Add Hangfire dashboard to consumer service
Hangfire has a dashboard that enables you to monitor the jobs and their statuses. It also allows you to manually trigger available jobs.
Open `Startup.cs` class and add Hangfire dependencies to `ConfigureServices` method:
```csharp


public void ConfigureServices(IServiceCollection services)
{
    services.AddControllers();
    services.AddHostedService<MessageReceiverHostedService>();

    services.AddHangfire(configuration => configuration
        .SetDataCompatibilityLevel(CompatibilityLevel.Version_170)
        .UseSimpleAssemblyNameTypeSerializer()
        .UseRecommendedSerializerSettings()
        .UseRedisStorage(_redis));
}

public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
{
    ...

    app.UseEndpoints(endpoints =>
    {
        endpoints.MapDefaultControllerRoute();
        endpoints.MapHangfireDashboard();
    });
}

Background jobs are processed by a [dedicated pool of worker](https://docs.hangfire.io/en/latest/background-processing/configuring-degree-of-parallelism.html) threads that run inside the Hangfire server subsystem. For each queue, we can config the number of workers.

- Step 13 - Add new class `HangfireQueueSetting` to bind configuration:
```csharp


internal class HangfireQueueSetting
{
    public string QueueName { get; set; }

    public int WorkerCount { get; set; }
}

public static void Main(string[] args)
{
    CreateHostBuilder(args).Build().Run();
}

public static IHostBuilder CreateHostBuilder(string[] args) =>
    Host.CreateDefaultBuilder(args)
        .ConfigureServices((hostContext, services) =>
        {
            _redis = ConnectionMultiplexer.Connect(hostContext.Configuration.GetConnectionString("RedisConnection"));

            services.AddHangfire(configuration => configuration
                .SetDataCompatibilityLevel(CompatibilityLevel.Version_170)
                .UseSimpleAssemblyNameTypeSerializer()
                .UseRecommendedSerializerSettings()
                .UseRedisStorage(_redis));

            var queueSettings = hostContext.Configuration.GetSection("Hangfire").Get<List<HangfireQueueSetting>>();
            foreach (var setting in queueSettings)
            {
                services.AddHangfireServer(options =>
                {
                    options.ServerName = $"{Environment.MachineName}:{setting.QueueName}";
                    options.Queues = new[] { setting.QueueName };
                    options.WorkerCount = setting.WorkerCount;
                });
            }
        });

- Step 16 - Reference shared class library project to producer service project

Now run all projects and navigate to Hangfire dashboard and you could see jobs are processing:
![Alt Text](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/916njnmii6sb6vqgsdu5.jpg)

Click on the `Servers` tab and you can see we have two queues and with different worker count:
![Alt Text](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/pl2hl21uhaoob0yh82mo.jpg)

####Dockerizing Services
```yml


version: '3.4'

networks:
  service_network:

services:
  redis:
    image: "redis"
    ports:
      - 6379:6379
    networks:
      - service_network

  consumerservice:
    image: ${DOCKER_REGISTRY-}jobqueueconsumerservice
    container_name: consumerservice
    ports:
      - 9000:80
    networks:
      - service_network
    build:
      context: .
      dockerfile: JobQueue.ConsumerService/Dockerfile
    environment:
      - ConnectionStrings__RedisConnection=redis:6379
      - JobApi__BaseAddress=http://producerservice

  producerservice:
    image: ${DOCKER_REGISTRY-}jobqueueproducerservice
    container_name: producerservice
    build:
      context: .
      dockerfile: JobQueue.ProducerService/Dockerfile
    networks:
      - service_network

  workerservice:
    image: ${DOCKER_REGISTRY-}workerservice
    networks:
      - service_network
    build:
      context: .
      dockerfile: JobQueue.WorkerService/Dockerfile
    environment:
      - ConnectionStrings__RedisConnection=redis:6379
      - Hangfire__0__WorkerCount=10
      - Hangfire__1__WorkerCount=5

Hangfire__0__WorkerCount=10