在本机相同进程中创建生产者消费者队列，可以解决很多线程安全以及高性能需求问题。本文告诉大家如何通过在 GitHub 完全开源的 AsyncWorkerCollection 库的 AsyncQueue 类创建在内存中的高性能低资源占用的生产者消费者队列

本文使用的 AsyncWorkerCollection 库在 GitHub 完全开源，请看 https://github.com/dotnet-campus/AsyncWorkerCollection/

这个库里面的所有代码都是在我团队实际项目经过约3年的测试，在大约 200 万台设备上运行过

本文使用的 AsyncQueue 是 AsyncWorkerCollection 库的其中一个类。通过 AsyncQueue 可以制作出一个支持多线程并发调用的生产者消费者队列，而 AsyncQueue 本身支持异步，可以配置线程池做到极低的资源占用

意义

为什么需要有生产者消费者队列？其实这个设计方法可以极大规避多个模块之间的性能差异，如下面例子

我有两个不同的模块，假定是 A 和 B 两个。业务逻辑要求让 A 模块执行完成的数据，进入到 B 模块。换句话说就是 B 模块的处理都需要依赖 A 模块的执行完成

但是现在存在的问题是 A 和 B 两个模块的执行速度有差异。如 A 模块是通过读取本机文件，而 B 模块是解析文件本身。或者说 A 模块是接收网络请求，而 B 模块是执行复杂的数据库逻辑

那么就会存在一个问题，能否让 A 和 B 独立执行，同时控制 A 和 B 两个模块的线程数量，以达到整体性能最佳？此时通过生产者消费者队列就能实现

按照如上描述，其实 A 模块就是生产者，生产出数据让 B 模块处理。而 B 模块就是消费者

如果 A 和 B 两个模块设计为独立执行，那么意味着可以让 A 和 B 两个模块的执行线程数量可以不匹配。如 A 的速度比较慢，此时分配给 A 更多的线程，或者说 B 执行比较慢分配给 B 更多的线程处理

在使用生产者消费者队列另一个意义在于可以做到资源的动态调配。如我有一个不重要的模块，这个模块需要处理一些杂务，而我不期望给这个模块投入太大的资源，但是我又期望在我应用空闲的时候可以将空闲资源投入处理。此时使用 生产者消费者队列 就能完成这个需求，所有模块将任务投入到生产者消费者队列里面，而平时只有很少的线程投入使用作为消费者处理这些任务。在应用空闲的时候，将更多线程投入到消费者处理里面处理

当然生产者消费者队列可以使用的业务将会很多，其他用途还请小伙伴自己摸索，或者百度一下

大部分的生产者消费者队列库都是设计为分布式的，支持多设备跨进程的，而这些库也就需要使用更多的资源。本文的 AsyncQueue 是在内存中创建，不会涉及到数据库等功能，只能在相同进程内使用。而因为没有跨进程和设备的功能，可以减少很多资源的时候，只需要一个简单的信号量锁就能完成

安装库

在使用之前的第一步就是安装 NuGet 库，本文的 AsyncWorkerCollection 库提供两个 NuGet 包，一个是 dll 引用，另一个是源代码引用，分别如下

• dotnetCampus.AsyncWorkerCollection
• dotnetCampus.AsyncWorkerCollection.Source

如果使用 SDK 版本的 csproj 可以在项目文件使用下面代码安装源代码版本

    <ItemGroup>
        <PackageReference Include="dotnetCampus.AsyncWorkerCollection.Source" Version="1.2.1">
            <PrivateAssets>all</PrivateAssets>
            <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
        </PackageReference>
    </ItemGroup>

使用方法

本质上的 AsyncQueue 的设计就是一个队列，因此用法和队列相同，有一个叫入队的方法，还有一个叫出队的方法。入队的方法是给生产者使用的，将数据或任务加入到队列里面提供给消费者使用。而出队就是给消费者使用的，消费者通过出队获取数据或任务用来执行

假定有数据是 FooTask 类，在创建 AsyncQueue 对象就需要传入对应的数据或任务的定义

    var asyncQueue = new AsyncQueue<FooTask>();

在生产者对应的代码里面使用 Enqueue 方法入队，这个方法是线程安全的，可以随意调用

   asyncQueue.Enqueue(new FooTask());

在消费者对应的代码里面使用 DequeueAsync 方法出队，这个方法是线程安全的，在队列里面没有数据的时候将会通过 await 等待，让线程返回线程池。在有数据的时候此方法将会返回。每一条数据都只会返回一次，也就是如果有多个线程同时调用 DequeueAsync 方法不会存在返回同一条数据。换句话说有多少次 Enqueue 方法入队，就会有多少次 DequeueAsync 的返回

    var fooTask = await asyncQueue.DequeueAsync();

使用方法就这么简单

那么如何用来做修改执行的线程的数量？如上面说的，本文的 AsyncQueue 是线程安全的，支持多个线程调用。因此如果修改调用 Enqueue 入队的线程数量就能修改生产者的线程数量。而修改等待 DequeueAsync 返回的线程数量也就能修改消费者的线程数量

也可以用来固定消费者的线程数量，用法很简单，就是预设置对应的消费者线程的数量就可以。如设置有三个线程进入循环等待 DequeueAsync 返回，那么消费者将最多使用三个线程执行。而在等待的时候通过 await 可以让线程返回线程池提升性能

有一点需要注意的是 AsyncQueue 是包含 Dispose 和 DisposeAsync 方法的，因为本质 AsyncQueue 使用了锁，需要手动释放。但主要需要调用的原因是让在业务完成之后，让没有手动释放的 DequeueAsync 方法可以释放，解决内存泄露问题。这部分原理请看 dotnet 使用 SemaphoreSlim 可能的内存泄露 大概的问题是在 AsyncQueue 对象不再使用的时候，如果有业务代码在 DequeueAsync 等待，那么这些业务代码引用的类将会存在内存泄露，不会被释放

在调用 Dispose 或 DisposeAsync 方法将可以释放 AsyncQueue 对象，在执行完成当前队列里面所有的数据之后，最后的 DequeueAsync 返回，解决内存泄露。而 Dispose 和 DisposeAsync 方法的不同在于，调用 DisposeAsync 方法相当于调用如下代码

    asyncQueue.Dispose();
    await asyncQueue.WaitForCurrentFinished();

例子

下面用一个例子让小伙伴了解使用的方法，和这个库的强大

假定有任务的定义如下

    class FooTask
    {
        public void Do()
        {
            Console.WriteLine("DoTask");
        }
    }

在实际使用的时候，其实更多将会定义为数据本身，让消费者执行相同的逻辑处理数据。而定义为任务本身可以提升灵活度，也就是每个任务可以使用不同的逻辑，但是需要小心任务本身的线程安全问题。如在 WPF 中不应该使用非 UI 线程访问 UI 线程控件等

这个任务有 100 个从线程池拿到的线程在创建，加入队列

            var random = new Random();
            var asyncQueue = new AsyncQueue<FooTask>();

            for (int i = 0; i < 100; i++)
            {
                Task.Run(async () =>
                {
                    while (true)
                    {
                        asyncQueue.Enqueue(new FooTask());

                        await Task.Delay(random.Next(1000));
                    }
                });
            }

而有对应的 10 个从线程池拿到的线程在执行

            for (int i = 0; i < 10; i++)
            {
                Task.Run(async () =>
                {
                    while (true)
                    {
                        var fooTask = await asyncQueue.DequeueAsync();
                        fooTask.Do();
                        Console.WriteLine($"剩余 {asyncQueue.Count}");
                        await Task.Delay(random.Next(50));
                    }
                });
            }

执行代码可以看到在消费线程里面将会不断从队列里面拿到任务，然后调用任务

本文代码放在github欢迎小伙伴访问

而 AsyncQueue 不仅可以用在多线程并发调用，也可以支持在单线程玩出协程的方法。先调用 DequeueAsync 加上等待，此时将会在当前线程注册等待调用，接着在其他业务模块调用入队的方法，每次调用入队将会回到出队的异步方法里面

如在 WPF 中添加下面代码

        private static async void DoTask(AsyncQueue<FooTask> asyncQueue)
        {
            while (true)
            {
                var task = await asyncQueue.DequeueAsync();
                task.Do();
                Console.WriteLine($"执行线程 {Thread.CurrentThread.ManagedThreadId}");
            }
        }

            var random = new Random();
            var asyncQueue = new AsyncQueue<FooTask>();

            DoTask(asyncQueue);

            for (int i = 0; i < 100; i++)
            {
                asyncQueue.Enqueue(new FooTask());

                await Task.Delay(random.Next(1000));
            }

可以看到里面代码全程没有创建线程也没有调用线程池，在入队之后将会在入队的线程调用到 await 出让才会让 DoTask 继续执行

如果代码不是在 WPF 中使用，而是在控制台就需要自己实现同步上下文，请看 C# dotnet 自己实现一个线程同步上下文

原理

其实这个 AsyncQueue 的本质就是使用 ConcurrentQueue 和 SemaphoreSlim 两个基础类创建的

关于 SemaphoreSlim 请看 C# dotnet 的锁 SemaphoreSlim 和队列

这个 SemaphoreSlim 锁的功能就是提供信号量，和异步等待的功能。信号量的用法就是设置多少次信号量就允许多少次使用信号量，这就是 AsyncQueue 可以让入队和出队的最大次数相等的原因

为什么是说最大次数而不是次数？原因是在于可以入队，但是没有线程调用 DequeueAsync 出队

在 DequeueAsync 方法底层调用的等待就是调用 SemaphoreSlim 的等待方法，如果没有信号量可以使用，那么这个等待将会等待到有信号量被设置。而等待是异步方法，也就是不会占用一个线程，此时占用的资源很小。当然用这个方法就需要小心 dotnet 使用 SemaphoreSlim 可能的内存泄露 这也就是在不使用 AsyncQueue 需要调用释放的原因

而 ConcurrentQueue 就是提供本身存放数据的类，这个类的设计就是线程安全的

因此通过 ConcurrentQueue 存放数据，而通过 SemaphoreSlim 通知出队，让 SemaphoreSlim 支持等待数据出队和让入队数量和出队最大数量相等

使用这两个类的配合就可以做到 AsyncQueue 的高性能低资源占用

本文使用的 AsyncWorkerCollection 库是完全开源基于 MIT 协议的库，欢迎小伙伴使用，在使用中遇到任何问题都欢迎在 GitHub 讨论

更多请看

• dotnet 使用 SemaphoreSlim 可能的内存泄露
• C# dotnet 的锁 SemaphoreSlim 和队列
• C# dotnet 自己实现一个线程同步上下文