这是一个 C++ 事件循环。在创建 objectEventLoop时，它会创建一个线程，该线程不断运行给它的任何任务。如果没有可用的任务，则主线程进入睡眠状态，直到添加了一些任务。
首先，我们需要一个线程安全队列，它允许多个生产者和至少一个消费者（EventLoop线程）。EventLoop控制消费者和生产者的对象。稍加改动，就可以添加多个消费者（runners 线程），而不是只添加一个线程。
#include <stdio.h>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <iostream>
#include <set>
#include <functional>
#if defined( WIN32 )
#include <windows.h>
#endif
class EventLoopNoElements : public std::runtime_error
{
public:
EventLoopNoElements(const char* error)
: std::runtime_error(error)
{
}
};
template <typename Type>
struct EventLoopCompare {
typedef std::tuple<std::chrono::time_point<std::chrono::system_clock>, Type> TimePoint;
bool operator()(const typename EventLoopCompare<Type>::TimePoint left, const typename EventLoopCompare<Type>::TimePoint right) {
return std::get<0>(left) < std::get<0>(right);
}
};
/**
* You can enqueue any thing with this event loop. Just use lambda functions, future and promises!
* With lambda `event.enqueue( 1000, [myvar, myfoo](){ myvar.something(myfoo); } )`
* With futures we can get values from the event loop:
* ```
* std::promise<int> accumulate_promise;
* event.enqueue( 2000, [&accumulate_promise](){ accumulate_promise.set_value(10); } );
* std::future<int> accumulate_future = accumulate_promise.get_future();
* accumulate_future.wait(); // It is not necessary to call wait, except for syncing the output.
* std::cout << "result=" << std::flush << accumulate_future.get() << std::endl;
* ```
* It is just not a nice ideia to add something which hang the whole event loop queue.
*/
template <class Type>
struct EventLoop {
typedef std::multiset<
typename EventLoopCompare<Type>::TimePoint,
EventLoopCompare<Type>
> EventLoopQueue;
bool _shutdown;
bool _free_shutdown;
std::mutex _mutex;
std::condition_variable _condition_variable;
EventLoopQueue _queue;
std::thread _runner;
// free_shutdown - if true, run all events on the queue before exiting
EventLoop(bool free_shutdown)
: _shutdown(false),
_free_shutdown(free_shutdown),
_runner( &EventLoop<Type>::_event_loop, this )
{
}
virtual ~EventLoop() {
std::unique_lock<std::mutex> dequeuelock(_mutex);
_shutdown = true;
_condition_variable.notify_all();
dequeuelock.unlock();
if (_runner.joinable()) {
_runner.join();
}
}
// Mutex and condition variables are not movable and there is no need for smart pointers yet
EventLoop(const EventLoop&) = delete;
EventLoop& operator =(const EventLoop&) = delete;
EventLoop(const EventLoop&&) = delete;
EventLoop& operator =(const EventLoop&&) = delete;
// To allow multiple threads to consume data, just add a mutex here and create multiple threads on the constructor
void _event_loop() {
while ( true ) {
try {
Type call = dequeue();
call();
}
catch (EventLoopNoElements&) {
return;
}
catch (std::exception& error) {
std::cerr << "Unexpected exception on EventLoop dequeue running: '" << error.what() << "'" << std::endl;
}
catch (...) {
std::cerr << "Unexpected exception on EventLoop dequeue running." << std::endl;
}
}
std::cerr << "The main EventLoop dequeue stopped running unexpectedly!" << std::endl;
}
// Add an element to the queue
void enqueue(int timeout, Type element) {
std::chrono::time_point<std::chrono::system_clock> timenow = std::chrono::system_clock::now();
std::chrono::time_point<std::chrono::system_clock> newtime = timenow + std::chrono::milliseconds(timeout);
std::unique_lock<std::mutex> dequeuelock(_mutex);
_queue.insert(std::make_tuple(newtime, element));
_condition_variable.notify_one();
}
// Blocks until getting the first-element or throw EventLoopNoElements if it is shutting down
// Throws EventLoopNoElements when it is shutting down and there are not more elements
Type dequeue() {
typename EventLoopQueue::iterator queuebegin;
typename EventLoopQueue::iterator queueend;
std::chrono::time_point<std::chrono::system_clock> sleeptime;
// _mutex prevents multiple consumers from getting the same item or from missing the wake up
std::unique_lock<std::mutex> dequeuelock(_mutex);
do {
queuebegin = _queue.begin();
queueend = _queue.end();
if ( queuebegin == queueend ) {
if ( _shutdown ) {
throw EventLoopNoElements( "There are no more elements on the queue because it already shutdown." );
}
_condition_variable.wait( dequeuelock );
}
else {
if ( _shutdown ) {
if (_free_shutdown) {
break;
}
else {
throw EventLoopNoElements( "The queue is shutting down." );
}
}
std::chrono::time_point<std::chrono::system_clock> timenow = std::chrono::system_clock::now();
sleeptime = std::get<0>( *queuebegin );
if ( sleeptime <= timenow ) {
break;
}
_condition_variable.wait_until( dequeuelock, sleeptime );
}
} while ( true );
Type firstelement = std::get<1>( *queuebegin );
_queue.erase( queuebegin );
dequeuelock.unlock();
return firstelement;
}
};
打印当前时间戳的实用程序：
std::string getTime() {
char buffer[20];
#if defined( WIN32 )
SYSTEMTIME wlocaltime;
GetLocalTime(&wlocaltime);
::snprintf(buffer, sizeof buffer, "%02d:%02d:%02d.%03d ", wlocaltime.wHour, wlocaltime.wMinute, wlocaltime.wSecond, wlocaltime.wMilliseconds);
#else
std::chrono::time_point< std::chrono::system_clock > now = std::chrono::system_clock::now();
auto duration = now.time_since_epoch();
auto hours = std::chrono::duration_cast< std::chrono::hours >( duration );
duration -= hours;
auto minutes = std::chrono::duration_cast< std::chrono::minutes >( duration );
duration -= minutes;
auto seconds = std::chrono::duration_cast< std::chrono::seconds >( duration );
duration -= seconds;
auto milliseconds = std::chrono::duration_cast< std::chrono::milliseconds >( duration );
duration -= milliseconds;
time_t theTime = time( NULL );
struct tm* aTime = localtime( &theTime );
::snprintf(buffer, sizeof buffer, "%02d:%02d:%02d.%03ld ", aTime->tm_hour, aTime->tm_min, aTime->tm_sec, milliseconds.count());
#endif
return buffer;
}
使用这些的示例程序：
// g++ -o test -Wall -Wextra -ggdb -g3 -pthread test.cpp && gdb --args ./test
// valgrind --leak-check=full --show-leak-kinds=all --track-origins=yes --verbose ./test
// procdump -accepteula -ma -e -f "" -x c:\ myexe.exe
int main(int argc, char* argv[]) {
std::cerr << getTime().c_str() << "Creating EventLoop" << std::endl;
EventLoop<std::function<void()>>* eventloop = new EventLoop<std::function<void()>>(true);
std::cerr << getTime().c_str() << "Adding event element" << std::endl;
eventloop->enqueue( 3000, []{ std::cerr << getTime().c_str() << "Running task 3" << std::endl; } );
eventloop->enqueue( 1000, []{ std::cerr << getTime().c_str() << "Running task 1" << std::endl; } );
eventloop->enqueue( 2000, []{ std::cerr << getTime().c_str() << "Running task 2" << std::endl; } );
std::this_thread::sleep_for( std::chrono::milliseconds(5000) );
delete eventloop;
std::cerr << getTime().c_str() << "Exiting after 10 seconds..." << std::endl;
return 0;
}
输出测试示例：
02:08:28.960 Creating EventLoop
02:08:28.960 Adding event element
02:08:29.960 Running task 1
02:08:30.961 Running task 2
02:08:31.961 Running task 3
02:08:33.961 Exiting after 10 seconds...