采取 BlockingCollection 的方法更慢？答案

【问题标题】：Take method of BlockingCollection slower?采取 BlockingCollection 的方法更慢？
【发布时间】：2020-07-24 11:53:56
【问题描述】：

我使用 BlockingCollection 对通过网络接收的数据进行排队并进行处理。我在计时器中每 100 毫秒消耗一次阻塞集合。但是我发现消费者很慢，阻塞收集的长度一直在增加。

由于我不太确定导致这种行为的原因，我编写了一个简单的控制台应用程序，如下所示。但是 BlockingCollection 计数仍然在不断增加。请注意，这次我使用了专用线程进行消费。

class Program
{
    static Thread sm_producerThread;
    static Thread sm_consumerThread;
    static BlockingCollection<int> sm_blockingCollection = new BlockingCollection<int>();
    static int sm_index;

    static void Main(string[] args)
    {
        sm_producerThread = new Thread(Produce) { IsBackground = true };
        sm_producerThread.Start();
        sm_consumerThread = new Thread(Consume) { IsBackground = true };
        sm_consumerThread.Start();
        Console.ReadLine();
    }
    
    private static void Produce()
    {
        while (true)
        {
            sm_index++;
            sm_blockingCollection.TryAdd(sm_index);
            Console.WriteLine($"Added number {sm_index}. Collection count is {sm_blockingCollection.Count}");
        }
    }

    private static void Consume()
    {
        while (true)
        {
            var num = sm_blockingCollection.Take();
            Console.WriteLine($"Fetched number {num}. Collection count is {sm_blockingCollection.Count}");
        }
    }
}

示例输出的片段如下所示

Fetched number 74317. Collection count is 1426
Fetched number 74318. Collection count is 1425
Fetched number 74319. Collection count is 1424
Fetched number 74320. Collection count is 1423
Fetched number 74321. Collection count is 1422
Fetched number 74322. Collection count is 1421
Fetched number 74323. Collection count is 1420
Added number 75743. Collection count is 1453
Added number 75744. Collection count is 1420
Added number 75745. Collection count is 1421
Added number 75746. Collection count is 1422
Added number 75747. Collection count is 1423
Added number 75748. Collection count is 1424
Added number 75749. Collection count is 1425
Added number 75750. Collection count is 1426
Added number 75751. Collection count is 1427
Added number 75752. Collection count is 1428
Added number 75753. Collection count is 1429
Added number 75754. Collection count is 1430
Added number 75755. Collection count is 1431
Added number 75756. Collection count is 1432
Added number 75757. Collection count is 1433
Added number 75758. Collection count is 1434
Added number 75759. Collection count is 1435
Added number 75760. Collection count is 1436
Added number 75761. Collection count is 1437
Added number 75762. Collection count is 1438
Added number 75763. Collection count is 1439
Fetched number 74324. Collection count is 1419
Added number 75764. Collection count is 1440
Added number 75765. Collection count is 1440
Added number 75766. Collection count is 1441
Added number 75767. Collection count is 1442
Added number 75768. Collection count is 1443
Added number 75769. Collection count is 1444
Added number 75770. Collection count is 1445
Added number 75771. Collection count is 1446
Added number 75772. Collection count is 1447
Added number 75773. Collection count is 1448
Added number 75774. Collection count is 1449
Added number 75775. Collection count is 1450
Added number 75776. Collection count is 1451
Added number 75777. Collection count is 1452
Added number 75778. Collection count is 1453
Added number 75779. Collection count is 1454
Added number 75780. Collection count is 1455
Added number 75781. Collection count is 1456
Added number 75782. Collection count is 1457
Added number 75783. Collection count is 1458
Added number 75784. Collection count is 1459
Added number 75785. Collection count is 1460
Added number 75786. Collection count is 1461
Added number 75787. Collection count is 1462
Added number 75788. Collection count is 1463
Added number 75789. Collection count is 1464
Added number 75790. Collection count is 1465
Added number 75791. Collection count is 1466
Added number 75792. Collection count is 1467
Added number 75793. Collection count is 1468
Added number 75794. Collection count is 1469
Added number 75795. Collection count is 1470
Added number 75796. Collection count is 1471
Added number 75797. Collection count is 1472
Added number 75798. Collection count is 1473
Added number 75799. Collection count is 1474
Added number 75800. Collection count is 1475
Added number 75801. Collection count is 1476
Added number 75802. Collection count is 1477
Added number 75803. Collection count is 1478
Added number 75804. Collection count is 1479
Added number 75805. Collection count is 1480
Added number 75806. Collection count is 1481
Added number 75807. Collection count is 1482
Added number 75808. Collection count is 1483
Added number 75809. Collection count is 1484
Added number 75810. Collection count is 1485
Added number 75811. Collection count is 1486
Added number 75812. Collection count is 1487
Added number 75813. Collection count is 1488
Added number 75814. Collection count is 1489
Added number 75815. Collection count is 1490
Added number 75816. Collection count is 1491
Added number 75817. Collection count is 1492
Added number 75818. Collection count is 1493

如您所见，在处理 75818 个条目时，计数实际上已达到 1490。我希望集合中的项目数在任何时候都很少（比如大约 100 个），因为两个线程可能具有几乎相同的优先级和相同的工作负载。我错过了什么？

【问题讨论】：

即使生产者每次只多创造 0.001% 的商品，消费者可以处理，平均而言，商品的数量也会增加。
如果您担心集合越来越大，请限制其大小 (docs.microsoft.com/en-us/dotnet/api/…)，然后使用 Add 而不是 TryAdd。这有效地增加了背压（即，如果消费者没有跟上，则减慢生产者的速度）。
您的实验没有显示Take 很慢。它表明Take 可能比TryAdd 稍微慢。它还表明，建立一个生产者-消费者系统，希望消费者至少与生产者一样快，这并不是构建该系统的稳健方式。
@mjwills 当我说工作量时，我并不是指“TryAdd”和采取方法的性能，而是剩余的代码块。但你的问题启发了我，我也应该考虑这些方面。
@TheodorZoulias 是的，我知道“慢”和“慢”这两个词之间有很多区别。更新了我的问题

标签： c# .net multithreading blockingcollection

【解决方案1】：

我复制了你的代码，将消费者线程的优先级提高到AboveNormal，并每1000个项目写入控制台。运行几秒钟后，我发现集合没有持续增加。我同意似乎暗示TryAdd 平均比Take 稍快的评论，如果线程以相同的优先级运行，则集合的大小会随着时间的推移而增加。

我每 100 毫秒从阻塞集合中消耗一次计时器。

如果我正确理解了这条评论，那么您正在使用计时器每秒 10 次唤醒您的处理线程。我是否假设您调用一次Take 然后等待下一个计时器事件？您的生产者多久发送一次数据？如果它每秒超过 10 次，而您每秒只处理 10 个数据块，那么问题是您没有经常为您的“缓冲区”（BlockingCollection）提供服务，这可以解释它的永久增长。

我使用的软件是一个美化的数据包嗅探器。当我们打开它时，我们会得到大量需要处理的网络数据包。我们的收集线程经过高度优化以促进这一点。下面是一些粗略的伪代码：

while (!shutdown)
{
    // Allocate the array for storing the packets.
    var packets = new packet[10000];

    // Start the timer using 100ms interval.
    timer.Start();

    // Read packets until the timer elapses or the capacity is reached.
    do
    {
        packets[index++] = device.Read(timeout);
    }
    while (!timerElapsed && packets.Length < capacity);

    // Notify the processing thread that data is available to be processed.
    PacketsReceived.Raise(this, new PacketEventArgs(packets));
}

除此之外还有更多内容，但这是基本要点。请注意，此线程仅专注于从设备接收数据。引发事件将数组传输到处理线程，然后使用ManualResetEvent 通知处理线程有数据可供处理。换句话说，处理线程处于空闲状态，直到收集线程通知它有工作要做，这作为数据交换的一部分发生。

使用您的示例，可以使用BlockingCollection 完成数据交换，以便收集线程将要处理的数据元素列表添加到集合中，同时处理线程从收藏。因此，当您的处理线程正在处理已接收的数据元素时，您的收集线程正在获取下一个要处理的数据元素列表。

HTH

【讨论】：