使用列表的消费者/生产者模式

Question

我有一个 WinForms 应用程序，其中包含一个消费者和一个生产者任务。我的生产者任务定期连接到 Web 服务并检索指定数量的字符串，然后需要将这些字符串放入某种并发的固定大小的 FIFO 队列中。然后我的消费者任务处理这些字符串，然后作为 SMS 消息发送出去（每条消息一个字符串）。我的生产者任务调用的 SOAP 函数需要一个参数来指定我想要获取的字符串的数量。这个数字将由我队列中的可用空间决定。因此，如果我的最大队列大小为 100 个字符串，并且下次我的生产者轮询 Web 服务时队列中有 60 个字符串，我需要它来请求 40 个字符串，因为那是我当时可以放入队列的所有内容.

这是我用来表示我的固定大小 FIFO 队列的代码：

public class FixedSizeQueue<T>
{
    private readonly List<T> queue = new List<T>();
    private readonly object syncObj = new object();

    public int Size { get; private set; }

    public FixedSizeQueue(int size)
    {
        Size = size;
    }

    public void Enqueue(T obj)
    {
        lock (syncObj)
        {
            queue.Insert(0, obj);

            if (queue.Count > Size)
            {
                queue.RemoveRange(Size, queue.Count - Size);
            }
        }
    }

    public T[] Dequeue()
    {
        lock (syncObj)
        {
            var result = queue.ToArray();
            queue.Clear();
            return result;
        }
    }

    public T Peek()
    {
        lock (syncObj)
        {
            var result = queue[0];
            return result;
        }
    }

    public int GetCount()
    {
        lock (syncObj)
        {
            return queue.Count;
        }
    }

我的生产者任务目前没有指定我需要从 Web 服务获得的字符串数量，但它似乎可以像获取队列中的当前项目计数 (q.GetCount()) 然后减去一样简单它来自我的最大队列大小。但是，即使 GetCount() 使用了锁，是否有可能一旦 GetCount() 退出，我的消费者任务就可以处理队列中的 10 个字符串，这意味着我永远无法真正保持队列 100%满了？

此外，我的消费者任务基本上需要先“窥视”队列中的第一个字符串，然后再尝试通过 SMS 消息发送它。如果无法发送消息，我需要将字符串保留在队列中的原始位置。我首先想到的是“偷看”队列中的第一个字符串，尝试在 SMS 消息中发送它，如果发送成功，然后将其从队列中删除。这样，如果发送失败，字符串仍然在队列中的原始位置。这听起来合理吗？

Answer 1

这是一个广泛的问题，所以确实没有明确的答案，但这是我的想法。

但是，即使 GetCount() 使用了锁，是不是只要 GetCount() 退出，我的消费者任务就可以处理队列中的 10 个字符串，这意味着我将永远无法真正保留队列 100% 已满？

是的，除非您在对 Web 服务的整个查询期间锁定 syncObj。但是生产者/消费者的重点是允许消费者在生产者获取更多物品的同时处理物品。你真的无能为力。在 某个 点，队列不会 100% 满。如果它总是 100% 满，那将意味着消费者根本没有做任何事情。

这样，如果发送失败，字符串仍然在队列中的原始位置。这听起来合理吗？

也许，但是按照您的编码方式，Dequeue() 操作会返回队列的整个状态并将其清除。给定此接口，您唯一的选择是重新排队失败的项目以供以后处理，这是一种非常合理的技术。

我还会考虑为消费者添加一种方法来阻止自己，直到有要处理的项目。例如：

public T[] WaitForItemAndDequeue(TimeSpan timeout)
{
    lock (syncObj) {
        if (queue.Count == 0 && !Monitor.Wait(syncObj, timeout)) {
            return null; // Timeout expired
        }

        return Dequeue();
    }
}

public T[] WaitForItem()
{
    lock (syncObj) {
        while (queue.Count != 0) {
            Monitor.Wait(syncObj);
        }

        return Dequeue();
    }
}

那么你必须在 Enqueue() 操作列表后更改为调用 Monitor.Pulse(syncObj)（所以在方法的末尾，但在 lock 块内）。