c++ - Windows C++ 线程等待队列数据推送

Question

我的程序设置如下:
有一个线程安全的队列类，一个线程在无限循环中向其推送数据，而第二个线程在无限循环中从队列中弹出数据。我正在尝试想出一种方法来使用 Windows 事件或其他一些机制来使 thread_1(如下)在无限 while 循环中等待，并且仅在队列深度大于或等于 1 时才迭代。

class thread-safe_Queue
{
 public:
  push();
  pop();
};

DWORD thread_1()
{
 while(1)
 {
  // wait for thread-safe queue to have data on it
  // pop data off
  // process data
 }
}

DWORD thread_2()
{
 while(1)
 {
  // when data becomes available, push data onto thread-safe queue
 }
}

Answer 1

我认为这可以解决问题。派生类 Event 并重载 Process() 函数。

#include <process.h> // Along with all the normal windows includes

//*********************************************
using namespace os;

Mutex globalQueueMutex;

class QueueReader : public Event
{
public:
    virtual void Process()
    {
      // Lock the queue
      Locker l(globalQueueMutex);
      // pop data off
      // process data
      return; // queue will automatically unlock
    }
};

QueueReader myQueueReader;

//*********************************************
// The queue writer would have functions like :
void StartQueueReader()
{
    Thread(QueueReader::StartEventHandler, &myQueueReader);
}
void WriteToQueue()
{
    Locker l(globalQueueMutex);
    // write to the queue
    myQueueReader.SignalProcess(); // tell reader to wake up
}
// When want to shutdown
void Shutdown()
{
    myQueueReader.SignalShutdown();
}

这里是执行魔法的类。

namespace os {

// **********************************************************************
/// Windows implementation to spawn a thread.
static uintptr_t Thread (void (*StartAddress)(void *), void *ArgList)
{
  return _beginthread(StartAddress, 0, ArgList);
}

// **********************************************************************
/// Windows implementation of a critical section.
class Mutex
{
public:
  // Initialize section on construction
  Mutex() { InitializeCriticalSection( &cs_ ); }
  // Delete section on destruction
  ~Mutex() { DeleteCriticalSection( &cs_ ); }
  // Lock it
  void lock() { EnterCriticalSection( &cs_ ); }
  // Unlock it
  void unlock() { LeaveCriticalSection( &cs_ ); }

private:
  CRITICAL_SECTION cs_;
}; // class Mutex

/// Locks/Unlocks a mutex
class Locker
{
public:
  // Lock the mutex on construction
  Locker( Mutex& mutex ): mutex_( mutex ) { mutex_.lock(); }
  // Unlock on destruction
  ~Locker() { mutex_.unlock(); }
private:
  Mutex& mutex_;
}; // class Locker

// **********************************************************************
// Windows implementation of event handler
#define ProcessEvent  hEvents[0]
#define SetTimerEvent hEvents[1]
#define ShutdownEvent hEvents[2]

/// Windows implementation of events
class Event
{
  /// Flag set when shutdown is complete
  bool Shutdown;
  /// Max time to wait for events
  DWORD Timer;
  /// The three events  - process, reset timer, and shutdown
  HANDLE hEvents[3];

public:
  /// Timeout is disabled by default and Events assigned
  Event( DWORD timer = INFINITE) : Timer(timer)
  {
    Shutdown = false;
    ProcessEvent = CreateEvent( NULL,TRUE,FALSE,NULL );
    SetTimerEvent = CreateEvent( NULL,TRUE,FALSE,NULL );
    ShutdownEvent = CreateEvent( NULL,TRUE,FALSE,NULL );
  }

  /// Close the event handles
  virtual ~Event()
  {
    CloseHandle(ProcessEvent);
    CloseHandle(SetTimerEvent);
    CloseHandle(ShutdownEvent);
  }

  /// os::Thread calls this to start the Event handler
  static void StartEventHandler(void *pMyInstance)
    { ((Event *)pMyInstance)->EventHandler(); }
  /// Call here to Change/Reset the timeout timer
  void ResetTimer(DWORD timer)  { Timer = timer; SetEvent(SetTimerEvent); }
  /// Set the signal to shutdown the worker thread processing events
  void SignalShutdown() { SetEvent(ShutdownEvent); while (!Shutdown) Sleep(30);}
  /// Set the signal to run the process
  void SignalProcess() { SetEvent(ProcessEvent); }

protected:
  /// Overload in derived class to process events with worker thread
  virtual void Process(){}
  /// Override to process timeout- return true to terminate thread
  virtual bool Timeout(){ return true;}

  /// Monitor thread events
  void EventHandler()
  {
    DWORD WaitEvents;
    while (!Shutdown)
    {
      // Wait here, looking to be signaled what to do next
      WaitEvents = WaitForMultipleObjects(3, hEvents, FALSE, Timer);

      switch (WaitEvents)
      {
        // Process event - process event then reset for the next one
        case WAIT_OBJECT_0 + 0:
          Process();
          ResetEvent(ProcessEvent);
          break;

        // Change timer event - see ResetTimer(DWORD timer)
        case WAIT_OBJECT_0 + 1:
          ResetEvent(SetTimerEvent);
          continue;

        // Shutdown requested so exit this thread
        case WAIT_OBJECT_0 + 2:
          Shutdown = true;
          break;

        // Timed out waiting for an event
        case WAIT_TIMEOUT:
          Shutdown = Timeout();
          break;

        // Failed - should never happen
        case WAIT_FAILED:
          break;

        default:
          break;
      }
    }
  }


};

} // namespace os