实用技巧

关注公众号 jb51net

关闭
首页 > 网络编程 > ASP.NET > 实用技巧 > .NET 缓冲队列 BufferQueue

.NET 高性能缓冲队列实现 BufferQueue的操作过程

作者:黑洞视界

BufferQueue 是一个用 .NET 编写的高性能的缓冲队列实现,支持多线程并发操作,这篇文章主要介绍了.NET 高性能缓冲队列实现 BufferQueue,需要的朋友可以参考下

前言

BufferQueue 是一个用 .NET 编写的高性能的缓冲队列实现,支持多线程并发操作。

项目是从 mocha 项目中独立出来的一个组件,经过修改以提供更通用的缓冲队列功能。

目前支持的缓冲区类型为内存缓冲区,后续会考虑支持更多类型的缓冲区。

适用场景

生产者和消费者之间的速度不一致,需要并发批量处理数据的场景。

功能说明

BufferQueue

需要注意的是,当前版本出于简化实现的考虑,暂不支持消费者的动态扩容和缩容,需要在创建消费者时指定消费者数量。

使用示例

安装 Nuget 包:

dotnet add package BufferQueue

项目基于 Microsoft.Extensions.DependencyInjection,使用时需要先注册服务。

BufferQueue 支持两种消费模式:pull 模式和 push 模式。

builder.Services.AddBufferQueue(options =>
{
    options.UseMemory(bufferOptions =>
        {
            // 每一对 Topic 和数据类型对应一个独立的缓冲区,可以设置 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模式 消费数据
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 容器中,可以通过构造函数注入其他服务,可以通过设置 ServiceLifetime 来控制 consumer 的生命周期。

BufferPushCustomerAttribute 中的 concurrency 参数用于设置 push consumer 的消费并发数,对应 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 有以下的特性:

BufferQueue

这个特性是通过以下两层接口设计实现的:

IBufferQueue

Partition 的设计

为了保证消费速度,BufferQueue 将数据划分为多个 Partition,每个 Partition 都是一个独立的队列,每个 Partition 都有一个对应的消费者线程。

Producer 以轮询的方式往每个 Partition 中写入数据。
Consumer 最多不允许超过 Partition 的数量,Partition 按平均分配到组内每个 Customer 上。
当一个 Consumer 被分配了多个 Partition 时,以轮训的方式进行消费。
每个 Partition 上会记录不同消费组的消费进度,不同组之间的消费进度互不干扰。

对并发的支持

Producer 支持并发写入。

Consumer 消费时是绑定 Partition 的,为保证能正确管理 Partition 的消费进度,Consumer 不支持并发消费。

如果要增加消费速度,需创建多个 Consumer。

Partition 的动态扩容

Partition 的基本组成单元是 Segment,Segment 代表保存数据的数组,多个 Segment 通过链表的形式组合成一个 Partition。

当一个 Segment 写满后,通过在其后面追加一个 Segment 实现扩容。

Segment 中用于保存数据的数组的每一个元素称为 Slot,每个 Slot 都有一个Partition 内唯一的自增 Offset。

Segment 的回收机制

每次在 Partition 中新增 Segment 时,会从头判断此前的 Segment 是否已经被所有消费组消费完,回收最后一个消费完的 Segment 作为新的 Segment 追加到 Partition 末尾使用。

Benchmark

测试环境:Apple M2 Max 64GB

写入性能测试

与 BlockingCollection 对比并发,并发线程数为 CPU 逻辑核心数 12, partitionNumber 为 1 和 12。

测试结果

Benchmark

在并发写入时,BufferQueue 的写入性能明显优于 BlockingCollection。

消费性能测试

pull 模式 consumer 与 BlockingCollection 对比并发读取性能,并发线程数为 CPU 逻辑核心数 12,partitionNumber 为 12。

测试结果

Benchmark

在批量消费时,随着批量大小的增加,BufferQueue 的消费性能优势更加明显。

到此这篇关于.NET 高性能缓冲队列实现 BufferQueue的文章就介绍到这了,更多相关.NET 高性能缓冲队列实现 BufferQueue内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

您可能感兴趣的文章:
阅读全文