ClickHouseAggregatingQueueProcessor

ClickHouseAggregatingQueueProcessor

ClickHouseAggregatingQueueProcessor was created to simplify the package aggregating and inserting to ClickHouse.

ClickHouseAggregatingQueueProcessor consists of:

• Processing queue with a set of worker-threads.
• The preparation queue to create the next package.

A package in a processing queue is a list of elements of an arbitrary type T (it is assumed that these will be entities for writing to ClickHouse).

Single elements are added to the preparation queue of the next package using the Add method. The package is created from the preparation queue and adds to the processing queue in the following cases:

1. The size of the next packet preparation queue exceeded MaxPackageSize.
2. By timer once in TimerPeriodMs ms.
3. When calling the PushPackageToQueue method.

Using the AddPackage method, you can add a package to the processing queue directly, bypassing the preparation queue.

When creating a queue, you must specify Action <IClickHouseRepository, List <T>, ILogger>, which will be executed by handler threads for each packet from the queue. The image below shows the scheme of working with packages.

---

Using with dependency injection

Add to DI

You can add ClickHouseAggregatingQueueProcessor into the DI mechanism can be done using the extension AddClickHouseAggregatingQueueProcessor<T>(IConfigurationSection configuration, Action <IClickHouseRepository, List <T>, ILogger> proc). Also you need to add to the DI ClickHouseRepository (see ClickHouseRepository wiki page).

ConfigureServices example code:

public void ConfigureServices(IServiceCollection services)
{       
    services.AddClickHouseRepository(config.GetSection("ClickHouseConnectionPoolConfiguration"));
    
    services.AddClickHouseAggregatingQueueProcessor(Configuration.GetSection("ClickHouseAggregatingQueueProcessorConfiguration"), proc);

    //...
}

private readonly Action<IClickHouseRepository, List<TestEntity>, ILogger> proc = (repository, package, logger) =>
{ 
    repository.BulkInsert(TestEntity.TableName, TestEntity.ColumnNames, package);
    logger.LogInformation($"Package inserted by worker-thread, package.Count: {package.Count}");
    // or other processing code 
};

In the extension method, you must pass Action for worker threads and the configuration section, which has the following form:

"ClickHouseAggregatingQueueProcessorConfiguration": {
"MaxPackageSize": 4,
"ProcessingThreadsCount": 2,
"QueueMaxSize": 10,
"TimerPeriodMs": 2000
},

Start ClickHouseAggregatingQueueProcessor

ClickHouseAggregatingQueueProcessor is added to the DI as Singleton. In addition to adding to the DI, you must also start it (by calling the Start method). You can do this yourself, or use the extension for IHost StartClickHouseAggregatingQueueProcessor in the Program class.

Example:

public class Program
{
    public static void Main(string[] args)
    {
        var host = CreateHostBuilder(args)
            .Build();
            .StartClickHouseAggregatingQueueProcessor<TestEntity>()
            .Run();
    }

    public static IHostBuilder CreateHostBuilder(string[] args) =>
        Host.CreateDefaultBuilder(args)
            .ConfigureWebHostDefaults(webBuilder =>
            {
                webBuilder.UseStartup<Startup>();
            });
}

Next, you can use ClickHouseAggregatingQueueProcessor:

public class TestClass
{
    private readonly IClickHouseAggregatingQueueProcessor<TestEntity> _queue;

    public TestClass(IClickHouseAggregatingQueueProcessor<TestEntity> queue)
    {
        _queue = queue;
    }

    public void AddSingle(TestEntity entity)
    {
        _queue.Add(entity);
    }
}