如何同步生产者线程和消费者线程 [C++/CLI]

Question

编辑：忘了说程序必须在 .NET 3.5 中； Visual Studio 2008 和 Windows 7。

在我的旧 MSDN 库（用于 Visual Studio 2008）文档中，我遇到了一篇关于 C# 中的线程的文章。我尝试转换为 C++/CLI。我在转换过程中遇到了很多问题。现在只剩下一个了：C3371: Cannot take the address of 'Producer::ThreadRun' unless creating delegate instance。

原始文档的标题：如何：同步生产者和消费者线程（C# 编程指南）

我在在线 MSDN 库中找不到原始文档。好像已经被删除了。我也无法在网络上的任何地方找到它。

在文档中，代码有：

• 在队列中保留 20 个整数的生产者线程
• 处理刚刚输入队列的数字的消费者线程
• 主线程将每 2.5 秒显示一次队列中的数字。当主线程在队列上工作时，其他线程被停止。

嗯，这里是 C# 的原始代码：

using System;
using System.Threading;
using System.Collections;
using System.Collections.Generic;

public class SyncEvents
{
    public SyncEvents()
    {

        _newItemEvent = new AutoResetEvent(false);
        _exitThreadEvent = new ManualResetEvent(false);
        _eventArray = new WaitHandle[2];
        _eventArray[0] = _newItemEvent;
        _eventArray[1] = _exitThreadEvent;
    }

    public EventWaitHandle ExitThreadEvent
    {
        get { return _exitThreadEvent; }
    }
    public EventWaitHandle NewItemEvent
    {
        get { return _newItemEvent; }
    }
    public WaitHandle[] EventArray
    {
        get { return _eventArray; }
    }

    private EventWaitHandle _newItemEvent;
    private EventWaitHandle _exitThreadEvent;
    private WaitHandle[] _eventArray;
}
public class Producer
{
    public Producer(Queue<int> q, SyncEvents e)
    {
        _queue = q;
        _syncEvents = e;
    }
    // Producer.ThreadRun
    public void ThreadRun()
    {
        int count = 0;
        Random r = new Random();
        while (!_syncEvents.ExitThreadEvent.WaitOne(0, false))
        {
            lock (((ICollection)_queue).SyncRoot)
            {
                while (_queue.Count < 20)
                {
                    _queue.Enqueue(r.Next(0, 100));
                    _syncEvents.NewItemEvent.Set();
                    count++;
                }
            }
        }
        Console.WriteLine("Producer thread: produced {0} items", count);
    }
    private Queue<int> _queue;
    private SyncEvents _syncEvents;
}

public class Consumer
{
    public Consumer(Queue<int> q, SyncEvents e)
    {
        _queue = q;
        _syncEvents = e;
    }
    // Consumer.ThreadRun
    public void ThreadRun()
    {
        int count = 0;
        while (WaitHandle.WaitAny(_syncEvents.EventArray) != 1)
        {
            lock (((ICollection)_queue).SyncRoot)
            {
                int item = _queue.Dequeue();
            }
            count++;
        }
        Console.WriteLine("Consumer Thread: consumed {0} items", count);
    }
    private Queue<int> _queue;
    private SyncEvents _syncEvents;
}

public class ThreadSyncSample
{
    private static void ShowQueueContents(Queue<int> q)
    {
        lock (((ICollection)q).SyncRoot)
        {
            foreach (int item in q)
            {
                Console.Write("{0, 2} ", item);
            }
        }
        Console.WriteLine();
    }

    static void Main()
    {
        Queue<int> queue = new Queue<int>();
        SyncEvents syncEvents = new SyncEvents();

        Console.WriteLine("Configuring worker threads...");
        Producer producer = new Producer(queue, syncEvents);
        Consumer consumer = new Consumer(queue, syncEvents);
        Thread producerThread = new Thread(producer.ThreadRun);
        Thread consumerThread = new Thread(consumer.ThreadRun);

        Console.WriteLine("Launching producer and consumer threads...");
        producerThread.Start();
        consumerThread.Start();

        for (int i = 0; i < 4; i++)
        {
            Thread.Sleep(2500);
            ShowQueueContents(queue);
        }

        Console.WriteLine("Signaling threads to terminate...");
        syncEvents.ExitThreadEvent.Set();

        producerThread.Join();
        consumerThread.Join();
    }
}

我已转换为以下 C++/CLI。请注意，

• 没有getter，因为我发现如果我有getter 我无法在Producer::ThreadRun 中获得WaitOne 函数
• 我在 C# 中使用 Monitor::Enter 和 Monitor::Exit，而不是 lock
• main() 在gcnew Thread(&Producer::ThreadRun) 创建线程时出现编译错误
• 错误：C3374：无法获取Producer::ThreadRun 的地址，除非创建委托实例。

C++/CLI 中的代码：

using namespace System;
using namespace System::Threading;
using namespace System::Collections;
using namespace System::Collections::Generic;

public ref class SyncEvents
{
public:
    property System::Threading::EventWaitHandle ^ _newItemEvent;
    property System::Threading::EventWaitHandle ^ _exitThreadEvent;
    property array<System::Threading::WaitHandle ^> ^ _eventArray;

public:
    SyncEvents()
    {
        _newItemEvent = gcnew System::Threading::AutoResetEvent(false);
        _exitThreadEvent = gcnew System::Threading::ManualResetEvent(false);

        _eventArray = gcnew array<System::Threading::WaitHandle ^>(2);
        _eventArray[0] = _newItemEvent;
        _eventArray[1] = _exitThreadEvent;
    }

};

public ref class Producer
{
private:
    System::Collections::Generic::Queue<int> ^ _queue;
    SyncEvents ^ _syncEvents;

public:
    Producer(System::Collections::Generic::Queue<int> ^ q, SyncEvents ^ e)
    {
        _queue = q;
        _syncEvents = e;
    }
    void ThreadRun()
    {
        int count = 0;
        Random ^ r = gcnew Random();

        while (!_syncEvents->_exitThreadEvent->WaitOne(0, false))
        {
            Monitor::Enter(((System::Collections::ICollection ^)_queue)->SyncRoot);
            try
            {
                while (_queue->Count < 20)
                {
                    _queue->Enqueue(r->Next(0, 100));
                    _syncEvents->_newItemEvent->Set();
                    count++;
                }
            }
            finally
            {
                Monitor::Exit(((System::Collections::ICollection ^)_queue)->SyncRoot);
            }
        }
        Console::WriteLine("Producer thread: produced {0} items", count);
    }
};

public ref class Consumer
{
private:
    System::Collections::Generic::Queue<int> ^ _queue;
    SyncEvents ^ _syncEvents;

public:
    Consumer(System::Collections::Generic::Queue<int> ^ q, SyncEvents ^ e)
    {
        _queue = q;
        _syncEvents = e;
    }
    void ThreadRun()
    {
        int count = 0;

        while (System::Threading::WaitHandle::WaitAny(_syncEvents->_eventArray) != 1)
        {
            Monitor::Enter(((System::Collections::ICollection ^)_queue)->SyncRoot);
            try
            {
                int item = _queue->Dequeue();
            }
            finally
            {
                Monitor::Exit(((System::Collections::ICollection ^)_queue)->SyncRoot);
            }
            count++;
        }
        Console::WriteLine("Consumer Thread: consumed {0} items", count);
    }
};

static void ShowQueueContents(System::Collections::Generic::Queue<int> ^ _q)
{
    Monitor::Enter(((System::Collections::ICollection ^)_q)->SyncRoot);
    try
    {
        for each (int item in _q)
            Console::WriteLine("{0, 3} ", item);
    }
    finally
    {
        Monitor::Exit(((System::Collections::ICollection ^)_q)->SyncRoot);
    }
};

int main()
{
    System::Collections::Generic::Queue<int> ^ queue 
        = gcnew System::Collections::Generic::Queue<int>(); 
    SyncEvents ^ syncEvents = gcnew SyncEvents();

    Console::WriteLine("Configuring worker threads...");
    Producer ^ producer = gcnew Producer(queue, syncEvents);
    Consumer ^ consumer = gcnew Consumer(queue, syncEvents);
    System::Threading::Thread ^ producerThread = gcnew System::Threading::Thread(&Producer::ThreadRun);
    System::Threading::Thread ^ consumerThread = gcnew System::Threading::Thread(&Consumer::ThreadRun);

    Console::WriteLine("Launching producer and cosumer threads...");
    producerThread->Start();
    consumerThread->Start();

    for (int i = 0; i < 4; i++)
    {
        Thread::Sleep(2500);
        ShoqQueueuContents(queue);
    }

    Console::WriteLine("Signaling threads to terminate...");        
    syncEvents->_exitThreadEvent->Set();

    producerThread->Join();
    consumerThread->Join();
}

Answer 1

 System::Threading::Thread ^ producerThread = 
      gcnew System::Threading::Thread(&Producer::ThreadRun);

Thread 类构造函数需要一个委托对象。您已经习惯了自动将方法转换为委托的 C# 语言语法糖。但是 C++/CLI 中不存在这种糖，您必须自己创建委托对象。并且要明确目标对象和目标方法，C# 编译器想要推断目标对象。这确实为 C++/CLI 语言提供了一些 C# 所没有的功能，但代价是必须编写更强大的代码：

 System::Threading::Thread ^ producerThread = 
      gcnew System::Threading::Thread(
          gcnew System::Threading::ThreadStart(producer, &Producer::ThreadRun)
      );

请考虑 .NET 4 中对这些场景的现有支持，例如 BlockingCollection 和 ConcurrentQueue。