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

windows c++ thread waiting on queue data-push

本文关键字：Windows c++ 线程 数据 等待队列      更新时间：2023-10-16

程序设置如下:
有一个线程安全的队列类，一个线程在无限循环中向它推送数据，另一个线程在无限循环中从它弹出数据。我正试图想出一种方法来使用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
 }
}

我想这可能会奏效。派生类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

这个怎么样(我假设您熟悉事件机制)

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);    
    }
  }

thread_2 

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

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

相关文章：