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

Answer 2

这个怎么样（我假设你熟悉事件机制）。

1.

thread_safe_Queue::push(something)
{
// lock the queue
...
// push object
// Signal the event
SetEvent(notification);

// unlock the queue
}

2.

thread_safe_Queue::pop(something)
{
WaitForSingleObject(notification);
// lock the queue
...
// get object
// reset the event
if (queue is empty)
  ResetEvent(notification);

// unlock the queue
}

3。 thread_1 只是尝试弹出对象并处理它。随着东西被推送，事件被启用，所以pop可以被成功调用。否则它将在pop 内等待。实际上，在这种情况下，您可以使用其他同步对象（如互斥锁或临界区）代替事件。

更新。外部事件： 线程 1：

void thread_1()
  {
  while(1)
    {
    WaitForSingleObject(notification);
    if (!pop(object))  // pop should return if there are any objects left in queue
      SetEvent(notification);    
    }
  }

线程_2

void thread_2()
  {
  while(1)
    {
    // push the object and than signal event
    ResetEvent(notification)
    }
  }

Answer 3

您可以使用命名事件。每个线程都会调用 CreateEvent 以相同的名称传递。然后使用 WaitForMultipleObjects 等待队列相关事件或结束程序事件。 pop 线程将等待 queue_has_data 和 end_program 事件。推送线程将等待 data_available 和 end_program 事件，并在将某些内容放入队列时设置 queue_has_data 事件。