在队列中添加数据时通知线程

本文关键字：通知 线程 数据 添加 队列 | 更新日期: 2023-09-27 18:31:50

我有一个线程正在队列中添加数据，现在我希望在添加数据时通知另一个线程，以便它可以开始处理队列中的数据。

一种选择是线程将不断轮询队列以查看计数是否大于零，但我认为这不是好方法，任何其他建议将不胜感激

任何关于如何实现这一目标的建议，我使用的是.net框架3.5。

如果我有两个线程，一个在做q.Enqueue(data)，另一个在做q.dequeue()，在这种情况下我需要管理锁..？

在队列中添加数据时通知线程

您可以使用

ManualResetEvent来通知线程。

ManualResetEvent e = new ManualResetEvent(false);

每次q.enqueue();执行e.Set()后，在处理线程中，您等待带有 e.WaitOne() 的项目。

如果在循环内进行处理，则应在e.WaitOne()后立即执行e.Reset()。

我不使用队列，因为我宁愿批处理它们。当您必须打开/关闭（日志）文件，打开/关闭数据库时，这更有用。这是我如何创建这样的示例：

// J. van Langen
public abstract class QueueHandler<T> : IDisposable
{
    // some events to trigger.
    ManualResetEvent _terminating = new ManualResetEvent(false);
    ManualResetEvent _terminated = new ManualResetEvent(false);
    AutoResetEvent _needProcessing = new AutoResetEvent(false);
    // my 'queue'
    private List<T> _queue = new List<T>();
    public QueueHandler()
    {
        new Thread(new ThreadStart(() =>
        {
            // what handles it should wait on.
            WaitHandle[] handles = new WaitHandle[] { _terminating, _needProcessing };
            // while not terminating, loop (0 timeout)
            while (!_terminating.WaitOne(0))
            {
                // wait on the _terminating and the _needprocessing handle.
                WaitHandle.WaitAny(handles);
                // my temporay array to store the current items.
                T[] itemsCopy;
                // lock the queue
                lock (_queue)
                {
                    // create a 'copy'
                    itemsCopy = _queue.ToArray();
                    // clear the queue.
                    _queue.Clear();
                }
                if (itemsCopy.Length > 0)
                    HandleItems(itemsCopy);
            }
            // the thread is done.
            _terminated.Set();
        })).Start();
    }
    public abstract void HandleItems(T[] items);
    public void Enqueue(T item)
    {
        // lock the queue to add the item.
        lock (_queue)
            _queue.Add(item);
        _needProcessing.Set();
    }
    // batch
    public void Enqueue(IEnumerable<T> items)
    {
        // lock the queue to add multiple items.
        lock (_queue)
            _queue.AddRange(items);
        _needProcessing.Set();
    }
    public void Dispose()
    {
        // let the thread know it should stop.
        _terminating.Set();
        // wait until the thread is stopped.
        _terminated.WaitOne();
    }
}

对于_terminating/_terminated，我使用ManualResetEvent，因为这些只是设置的。

对于_needProcessing，我使用AutoResetEvent它不能用 ManualResetEvent 来完成，因为当它被触发时，另一个线程可能会再次Set它，所以如果你在 WaitHandle.WaitAny 之后Reset它，你可以撤消新添加的项目。（嗯，如果有人能更容易地解释这一点，欢迎:)

例：

public class QueueItem
{
}
public class MyQueue : QueueHandler<QueueItem>
{
    public override void HandleItems(QueueItem[] items)
    {
        // do your thing.
    }
}

public void Test()
{
    MyQueue queue = new MyQueue();
    QueueItem item = new QueueItem();
    queue.Enqueue(item);
    QueueItem[] batch = new QueueItem[]
    {
        new QueueItem(),
        new QueueItem()
    };
    queue.Enqueue(batch);
    // even on dispose, all queued items will be processed in order to stop the QueueHandler.
    queue.Dispose();
}

使用 BlockingCollection 类。这样做的好处是，如果队列为空，则Take方法会阻止（不轮询）。它包含在 .NET 4.0+ 中，或者作为反应式扩展下载的一部分，甚至可能通过 NuGet 向后移植 TPL。如果需要，可以使用类的以下未优化变体。

public class BlockingCollection<T>
{
    private readonly Queue<T> m_Queue = new Queue<T>();
    public void Add(T item)
    {
        lock (m_Queue)
        {
            m_Queue.Enqueue(item);
            Monitor.Pulse(m_Queue);
        }
    }
    public T Take()
    {
        lock (m_Queue)
        {
            while (m_Queue.Count == 0)
            {
                Monitor.Wait(m_Queue);
            }
            return m_Queue.Dequeue();
        }
    }
    public bool TryTake(out T item)
    {
        item = default(T);
        lock (m_Queue)
        {
            if (m_Queue.Count > 0)
            {
                item = m_Queue.Dequeue();
            }
        }
        return item != null;
    }
}

我认为BlockingCollection会比Queue做得更好。除此之外，不断检查队列大小（并在线程为零时暂停线程）是相当不错的方法。

顺便说一句，我们在这里谈论的是生产者 - 消费者模式。我想你可以用谷歌搜索其他一些方法。

相关文章：