.NET?高性能緩沖隊列實現(xiàn)?BufferQueue的操作過程

更新時間：2024年07月30日 08:58:51 作者：黑洞視界

BufferQueue 是一個用 .NET 編寫的高性能的緩沖隊列實現(xiàn),支持多線程并發(fā)操作,這篇文章主要介紹了.NET?高性能緩沖隊列實現(xiàn)?BufferQueue,需要的朋友可以參考下

前言

BufferQueue 是一個用 .NET 編寫的高性能的緩沖隊列實現(xiàn)，支持多線程并發(fā)操作。

項目是從 mocha 項目中獨立出來的一個組件，經過修改以提供更通用的緩沖隊列功能。

目前支持的緩沖區(qū)類型為內存緩沖區(qū)，后續(xù)會考慮支持更多類型的緩沖區(qū)。

適用場景

生產者和消費者之間的速度不一致，需要并發(fā)批量處理數(shù)據的場景。

功能說明

支持創(chuàng)建多個 Topic，每個 Topic 可以有多種數(shù)據類型。每一對 Topic 和數(shù)據類型對應一個獨立的緩沖區(qū)。

BufferQueue

支持創(chuàng)建多個 Consumer Group，每個 Consumer Group 的消費進度都是獨立的。支持多個 Consumer Group 并發(fā)消費同一個 Topic。
支持同一個 Consumer Group 創(chuàng)建多個 Consumer，以負載均衡的方式消費數(shù)據。
支持數(shù)據的批量消費，可以一次性獲取多條數(shù)據。
支持 pull 模式和 push 模式兩種消費模式。
pull 模式下和 push 模式下都支持 auto commit 和 manual commit 兩種提交方式。auto commit 模式下，消費者在收到數(shù)據后自動提交消費進度，如果消費失敗不會重試。manual commit 模式下，消費者需要手動提交消費進度，如果消費失敗只要不提交進度就可以重試。

需要注意的是，當前版本出于簡化實現(xiàn)的考慮，暫不支持消費者的動態(tài)擴容和縮容，需要在創(chuàng)建消費者時指定消費者數(shù)量。

使用示例

安裝 Nuget 包：

dotnet add package BufferQueue

項目基于 Microsoft.Extensions.DependencyInjection，使用時需要先注冊服務。

BufferQueue 支持兩種消費模式：pull 模式和 push 模式。

builder.Services.AddBufferQueue(options =>
{
    options.UseMemory(bufferOptions =>
        {
            // 每一對 Topic 和數(shù)據類型對應一個獨立的緩沖區(qū)，可以設置 partitionNumber
            bufferOptions.AddTopic<Foo>("topic-foo1", partitionNumber: 6);
            bufferOptions.AddTopic<Foo>("topic-foo2", partitionNumber: 4);
            bufferOptions.AddTopic<Bar>("topic-bar", partitionNumber: 8);
        })
        // 添加 push 模式的消費者
        // 掃描指定程序集中的標記了 BufferPushCustomerAttribute 的類，
        // 注冊為 push 模式的消費者
        .AddPushCustomers(typeof(Program).Assembly);
});
// 在 HostedService 中使用 pull模式 消費數(shù)據
builder.Services.AddHostedService<Foo1PullConsumerHostService>();

pull 模式的消費者示例：

public class Foo1PullConsumerHostService(
    IBufferQueue bufferQueue,
    ILogger<Foo1PullConsumerHostService> logger) : IHostedService
{
    private readonly CancellationTokenSource _cancellationTokenSource = new();
    public Task StartAsync(CancellationToken cancellationToken)
    {
        var token = CancellationTokenSource
            .CreateLinkedTokenSource(cancellationToken, _cancellationTokenSource.Token)
            .Token;
        var consumers = bufferQueue.CreatePullConsumers<Foo>(
            new BufferPullConsumerOptions
            {
                TopicName = "topic-foo1", GroupName = "group-foo1", AutoCommit = true, BatchSize = 100,
            }, consumerNumber: 4);
        foreach (var consumer in consumers)
        {
            _ = ConsumeAsync(consumer, token);
        }
        return Task.CompletedTask;
    }
    public Task StopAsync(CancellationToken cancellationToken)
    {
        _cancellationTokenSource.Cancel();
        return Task.CompletedTask;
    }
    private async Task ConsumeAsync(IBufferPullConsumer<Foo> consumer, CancellationToken cancellationToken)
    {
        await foreach (var buffer in consumer.ConsumeAsync(cancellationToken))
        {
            foreach (var foo in buffer)
            {
                // Process the foo
                logger.LogInformation("Foo1PullConsumerHostService.ConsumeAsync: {Foo}", foo);
            }
        }
    }
}

push 模式的消費者示例：

通過 BufferPushCustomer 特性注冊 push 模式的消費者。

push consumer 會被注冊到 DI 容器中，可以通過構造函數(shù)注入其他服務，可以通過設置 ServiceLifetime 來控制 consumer 的生命周期。

BufferPushCustomerAttribute 中的 concurrency 參數(shù)用于設置 push consumer 的消費并發(fā)數(shù)，對應 pull consumer 的 consumerNumber。

[BufferPushCustomer(
    topicName: "topic-foo2",
    groupName: "group-foo2",
    batchSize: 100,
    serviceLifetime: ServiceLifetime.Singleton,
    concurrency: 2)]
public class Foo2PushConsumer(ILogger<Foo2PushConsumer> logger) : IBufferAutoCommitPushConsumer<Foo>
{
    public Task ConsumeAsync(IEnumerable<Foo> buffer, CancellationToken cancellationToken)
    {
        foreach (var foo in buffer)
        {
            logger.LogInformation("Foo2PushConsumer.ConsumeAsync: {Foo}", foo);
        }
        return Task.CompletedTask;
    }
}

[BufferPushCustomer(
    "topic-bar",
    "group-bar",
    100,
    ServiceLifetime.Scoped,
    2)]
public class BarPushConsumer(ILogger<BarPushConsumer> logger) : IBufferManualCommitPushConsumer<Bar>
{
    public async Task ConsumeAsync(IEnumerable<Bar> buffer, IBufferConsumerCommitter committer,
        CancellationToken cancellationToken)
    {
        foreach (var bar in buffer)
        {
            logger.LogInformation("BarPushConsumer.ConsumeAsync: {Bar}", bar);
        }
        var commitTask = committer.CommitAsync();
        if (!commitTask.IsCompletedSuccessfully)
        {
            await commitTask.AsTask();
        }
    }
}

BufferQueue 內部設計概述

Topic 的隔離

BufferQueue 有以下的特性：

同一個數(shù)據類型下的不同 Topic 的 BufferQueue 互不干擾。
同一個 Topic 下的不同數(shù)據類型的 BufferQueue 互不干擾。

BufferQueue

這個特性是通過以下兩層接口設計實現(xiàn)的：

IBufferQueue：根據 TopicName 和 類型參數(shù) T 將請求轉發(fā)給具體的 IBufferQueue<T> 實現(xiàn)（借助 KeyedService 實現(xiàn)），其中參數(shù) T 代表 Buffer 所承載的數(shù)據實體的類型。
IBufferQueue<T>：具體的 BufferQueue 實現(xiàn)，負責管理 Topic 下的數(shù)據。屬于 Buffer 模塊的內部實現(xiàn)，不對外暴露。

IBufferQueue

Partition 的設計

為了保證消費速度，BufferQueue 將數(shù)據劃分為多個 Partition，每個 Partition 都是一個獨立的隊列，每個 Partition 都有一個對應的消費者線程。

Producer 以輪詢的方式往每個 Partition 中寫入數(shù)據。
Consumer 最多不允許超過 Partition 的數(shù)量，Partition 按平均分配到組內每個 Customer 上。
當一個 Consumer 被分配了多個 Partition 時，以輪訓的方式進行消費。
每個 Partition 上會記錄不同消費組的消費進度，不同組之間的消費進度互不干擾。