Как мне написать код производителя / потребителя в C #?

Question

У меня есть 1 поток потоковых данных и 2 (поток потоков), обрабатывающий данные.Обработка данных занимает около 100 мс, поэтому я использую второй поток, чтобы не удерживать 1-й поток.

Пока 2-й поток обрабатывает данные, 1-й поток добавляет данные в кэш словаря, а затем, когда 2-й потокПосле завершения он обрабатывает кэшированные значения.

Мои вопросы: как делать код производителя / потребителя в C #?

public delegate void OnValue(ulong value);

public class Runner
{
    public event OnValue OnValueEvent;
    private readonly IDictionary<string, ulong> _cache = new Dictionary<string, ulong>(StringComparer.InvariantCultureIgnoreCase);
    private readonly AutoResetEvent _cachePublisherWaitHandle = new AutoResetEvent(true);

    public void Start()
    {
        for (ulong i = 0; i < 500; i++)
        {
            DataStreamHandler(i.ToString(), i);
        }
    }

    private void DataStreamHandler(string id, ulong value)
    {
        _cache[id] = value;

        if (_cachePublisherWaitHandle.WaitOne(1))
        {
            IList<ulong> tempValues = new List<ulong>(_cache.Values);
            _cache.Clear();

            _cachePublisherWaitHandle.Reset();

            ThreadPool.UnsafeQueueUserWorkItem(delegate
            {
                try
                {
                    foreach (ulong value1 in tempValues)
                        if (OnValueEvent != null)
                            OnValueEvent(value1);
                }
                finally
                {
                    _cachePublisherWaitHandle.Set();
                }
            }, null);
        }
        else
        {
            Console.WriteLine(string.Format("Buffered value: {0}.", value));
        }
    }
}

class Program
{
    static void Main(string[] args)
    {
        Stopwatch sw = Stopwatch.StartNew();
        Runner r = new Runner();
        r.OnValueEvent += delegate(ulong x)
                              {
                                  Console.WriteLine(string.Format("Processed value: {0}.", x));
                                  Thread.Sleep(100);

                                  if(x == 499)
                                  {
                                      sw.Stop();
                                      Console.WriteLine(string.Format("Time: {0}.", sw.ElapsedMilliseconds));
                                  }
                              };
        r.Start();
        Console.WriteLine("Done");
        Console.ReadLine();
    }
}

Answer 1

Лучшей практикой для настройки шаблона производитель-потребитель является использование класса BlockingCollection , доступного в .NET 4.0, или отдельная загрузка структуры Reactive Extensions . Идея состоит в том, что производители будут ставить в очередь, используя метод Add, а потребители - в очередь, используя метод Take, который блокирует, если очередь пуста. Как SwDevMan81 указал, что на сайте Albahari действительно хорошая статья о том, как заставить его работать правильно, если вы хотите идти по ручному маршруту.

Answer 2

На MSDN есть хорошая статья о синхронизации источника и потребителя. Есть также хороший пример на сайте Albahari .