这两天天气凉了，苏州这边连续好几天都是淅淅沥沥的下着小雨，今天天气还稍微好点。前两天早上起来突然就感冒了，当天就用了一卷纸，好在年轻扛得住，第二天就跟没事人似的。在这里提醒大家一下，天气凉了，睡凉席的可以收起来了，体质不太好的，也要适当加点衣服。
本来是想接着看源码的，早上起来又把Emitter鼓捣了一下，跟大家说说。
emitter.hpp是可以从源码中剥离出来的，只要去除里面的libuv的东西就行了。Emitter其实就是实现的即时回调，没有异步事件处理的功能。但是我们有时候是需要用并发来提高处理速度的，于是我就把Emitter稍微改造了一下，先上代码：
#pragma once #include <type_traits>
#include <functional>
#include <algorithm>
#include <utility>
#include <cstddef>
#include <vector>
#include <memory>
#include <list>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <chrono> template<typename E>
using event_ptr = std::unique_ptr<E>; template<typename E, typename... Args>
event_ptr<E> make_event(Args&&... args) {
return std::make_unique<E>(std::forward<Args>(args)...);
} /**
* @brief Event emitter base class.
*/
template<typename T>
class Emitter : public std::enable_shared_from_this<T> {
struct BaseHandler {
virtual ~BaseHandler() noexcept = default;
virtual bool empty() const noexcept = ;
virtual void clear() noexcept = ;
virtual void join() noexcept = ;
virtual void exit() noexcept = ;
}; template<typename E>
struct Handler final: BaseHandler {
using Listener = std::function<void(E &, std::shared_ptr<T>&)>;
using Element = std::pair<bool, Listener>;
using ListenerList = std::list<Element>;
using Connection = typename ListenerList::iterator; bool empty() const noexcept override {
auto pred = [](auto &&element){ return element.first; }; return std::all_of(onceL.cbegin(), onceL.cend(), pred) &&
std::all_of(onL.cbegin(), onL.cend(), pred);
} void clear() noexcept override {
if(!publishing.try_lock()) {
auto func = [](auto &&element){ element.first = true; };
std::for_each(onceL.begin(), onceL.end(), func);
std::for_each(onL.begin(), onL.end(), func);
} else {
onceL.clear();
onL.clear();
}
publishing.unlock();
} Connection once(Listener f) {
return onceL.emplace(onceL.cend(), false, std::move(f));
} Connection on(Listener f) {
return onL.emplace(onL.cend(), false, std::move(f));
} void erase(Connection conn) noexcept {
conn->first = true; if(publishing.try_lock()) {
auto pred = [](auto &&element){ return element.first; };
onceL.remove_if(pred);
onL.remove_if(pred);
}
publishing.unlock();
} void run(E event, std::shared_ptr<T> ptr) {
ListenerList currentL;
onceL.swap(currentL); auto func = [&event, &ptr](auto &&element) {
return element.first ? void() : element.second(event, ptr);
}; publishing.lock(); std::for_each(onL.rbegin(), onL.rend(), func);
std::for_each(currentL.rbegin(), currentL.rend(), func); publishing.unlock(); onL.remove_if([](auto &&element){ return element.first; });
} void thread_fun(std::shared_ptr<T> ptr)
{
while(true) {
std::unique_lock<std::mutex> lk(emutex);
econd.wait_for(lk, std::chrono::milliseconds(), [this](){return !events.empty();}); if(events.size() > ) {
E event = std::move(events.front());
events.pop();
run(std::move(event), std::move(ptr));
} else {
break;
}
}
} void publish(E event, std::shared_ptr<T> ptr, bool asyn) {
if(asyn) {
{
std::lock_guard<std::mutex> lk(emutex);
events.push(std::move(event));
}
econd.notify_all(); if(!ethread.joinable()) {
ethread = std::thread(&Handler<E>::thread_fun, this, std::move(ptr));
}
} else {
run(std::move(event), ptr);
}
} void join() noexcept override {
if(ethread.joinable()) {
ethread.join();
}
} void exit() noexcept override {
if(ethread.joinable()) {
econd.notify_all();
ethread.join();
}
} private:
std::mutex publishing;
ListenerList onceL{};
ListenerList onL{}; std::thread ethread;
std::queue<E> events;
std::mutex emutex;
std::condition_variable econd;
}; static std::size_t next_type() noexcept {
static std::size_t counter = ;
return counter++;
} template<typename>
static std::size_t event_type() noexcept {
static std::size_t value = next_type();
return value;
} template<typename E>
Handler<E> & handler() noexcept {
std::size_t type = event_type<E>(); if(!(type < handlers.size())) {
handlers.resize(type+);
} if(!handlers[type]) {
handlers[type] = std::make_unique<Handler<E>>();
} return static_cast<Handler<E>&>(*handlers[type]);
} protected:
template<typename E>
void publish(E event, bool asyn = false) {
// handler<E>().publish(std::move(event), *static_cast<T*>(this), asyn);
handler<E>().publish(std::move(event), this->shared_from_this(), asyn);
} public:
template<typename E>
using Listener = typename Handler<E>::Listener; /**
* @brief Connection type for a given event type.
*
* Given an event type `E`, `Connection<E>` is the type of the connection
* object returned by the event emitter whenever a listener for the given
* type is registered.
*/
template<typename E>
struct Connection: private Handler<E>::Connection {
template<typename> friend class Emitter; Connection() = default;
Connection(const Connection &) = default;
Connection(Connection &&) = default; Connection(typename Handler<E>::Connection conn)
: Handler<E>::Connection{std::move(conn)}
{} Connection & operator=(const Connection &) = default;
Connection & operator=(Connection &&) = default;
}; virtual ~Emitter() noexcept {
static_assert(std::is_base_of<Emitter<T>, T>::value, "!");
} /**
* @brief Registers a long-lived listener with the event emitter.
*
* This method can be used to register a listener that is meant to be
* invoked more than once for the given event type.<br/>
* The Connection object returned by the method can be freely discarded. It
* can be used later to disconnect the listener, if needed.
*
* Listener is usually defined as a callable object assignable to a
* `std::function<void(const E &, T &)`, where `E` is the type of the event
* and `T` is the type of the resource.
*
* @param f A valid listener to be registered.
* @return Connection object to be used later to disconnect the listener.
*/
template<typename E>
Connection<E> on(Listener<E> f) {
return handler<E>().on(std::move(f));
} /**
* @brief Registers a short-lived listener with the event emitter.
*
* This method can be used to register a listener that is meant to be
* invoked only once for the given event type.<br/>
* The Connection object returned by the method can be freely discarded. It
* can be used later to disconnect the listener, if needed.
*
* Listener is usually defined as a callable object assignable to a
* `std::function<void(const E &, T &)`, where `E` is the type of the event
* and `T` is the type of the resource.
*
* @param f Avalid listener to be registered.
* @return Connection object to be used later to disconnect the listener.
*/
template<typename E>
Connection<E> once(Listener<E> f) {
return handler<E>().once(std::move(f));
} /**
* @brief Disconnects a listener from the event emitter.
* @param conn A valid Connection object
*/
template<typename E>
void erase(Connection<E> conn) noexcept {
handler<E>().erase(std::move(conn));
} /**
* @brief Disconnects all the listeners for the given event type.
*/
template<typename E>
void clear() noexcept {
handler<E>().clear();
} /**
* @brief Disconnects all the listeners.
*/
void clear() noexcept {
std::for_each(handlers.begin(), handlers.end(),
[](auto &&hdlr){ if(hdlr) { hdlr->clear(); } });
} /**
* @brief Checks if there are listeners registered for the specific event.
* @return True if there are no listeners registered for the specific event,
* false otherwise.
*/
template<typename E>
bool empty() const noexcept {
std::size_t type = event_type<E>(); return (!(type < handlers.size()) ||
!handlers[type] ||
static_cast<Handler<E>&>(*handlers[type]).empty());
} /**
* @brief Checks if there are listeners registered with the event emitter.
* @return True if there are no listeners registered with the event emitter,
* false otherwise.
*/
bool empty() const noexcept {
return std::all_of(handlers.cbegin(), handlers.cend(),
[](auto &&hdlr){ return !hdlr || hdlr->empty(); });
} void thread_join() const noexcept {
std::for_each(handlers.begin(), handlers.end(),
[](auto &&hdlr){ if(hdlr) { hdlr->join(); } });
} void thread_exit() const noexcept {
std::for_each(handlers.begin(), handlers.end(),
[](auto &&hdlr){ if(hdlr) { hdlr->exit(); } });
} private:
std::vector<std::unique_ptr<BaseHandler>> handlers{};
};
emitter.h
Emitter类应该和项目是兼容的，但是为了更干净和通用一点，去除了ErrorEvent这些东西，所以已经不适合再放到源码里。下面是使用的例子：
#include <iostream>
#include <memory>
#include <thread> using namespace std;
#include "emitter.h" struct StringEvent
{
StringEvent(std::string str):i_str(str)
{
cout << "StringEvent" << std::endl;
} void print()
{
std::cout << "string event:" << i_str << std::endl;
} std::string i_str; ~StringEvent()
{
cout << "~StringEvent" << std::endl;
}
}; struct IntEvent
{
IntEvent(int t) : i_t(t)
{
cout << "IntEvent" << std::endl;
}
void print()
{
std::cout << "int event:" << i_t << std::endl;
}
~IntEvent()
{
cout << "~IntEvent" << std::endl;
} int i_t{};
}; class A : public Emitter<A>
{
public:
A()
{
cout << "A" << endl;
} void print()
{
publish(StringEvent("Hello"), false);
publish(make_event<StringEvent>("Hello"), true);
publish(make_unique<StringEvent>("World"), true); this_thread::sleep_for(1000ms);
publish(make_unique<IntEvent>(), true);
publish(make_unique<IntEvent>(), true); } ~A()
{
cout << "~A" << endl;
}
}; int main()
{
shared_ptr<A> em = make_shared<A>(); em->on<StringEvent>([](StringEvent& ev, shared_ptr<A>& a){
ev.print();
}); em->on<event_ptr<StringEvent>>([](event_ptr<StringEvent>& ev, shared_ptr<A>& a){
ev->print();
}); em->on<event_ptr<IntEvent>>([](event_ptr<IntEvent>& ev, shared_ptr<A>& a){
ev->print();
}); em->print(); em->thread_join(); return ;
}
test.cc
主要来看看做的一些改动。
先看test.cc里面，现在的事件处理函数Lambda中的两个参数做了改变：
em->on<StringEvent>([](StringEvent& ev, shared_ptr<A>& a){
ev.print();
});
第二个参数由原来的 A& 类型 改成了 share_ptr<A>& 类型。
class A : public Emitter<A>
{
public:
A()
{
cout << "A" << endl;
} void print()
{
publish(StringEvent("Hello"), false);
publish(make_event<StringEvent>("Hello"), true);
publish(make_unique<StringEvent>("World"), true); this_thread::sleep_for(1000ms);
publish(make_unique<IntEvent>(), true);
publish(make_unique<IntEvent>(), true); } ~A()
{
cout << "~A" << endl;
}
};
我们看过之前的代码，现在在publish中多加了一个bool参数，默认值为false，用于指示这个事件是否需要异步处理。
这里注意下，指示异步处理的是在发生事件，调用publish的时候。
用法上面主要就这些改动，然后再来看emitter.h
Handler类
template<typename E>
struct Handler final: BaseHandler {
using Listener = std::function<void(E &, std::shared_ptr<T>&)>;
using Element = std::pair<bool, Listener>;
using ListenerList = std::list<Element>;
using Connection = typename ListenerList::iterator; bool empty() const noexcept override {
auto pred = [](auto &&element){ return element.first; }; return std::all_of(onceL.cbegin(), onceL.cend(), pred) &&
std::all_of(onL.cbegin(), onL.cend(), pred);
} void clear() noexcept override {
if(!publishing.try_lock()) {
auto func = [](auto &&element){ element.first = true; };
std::for_each(onceL.begin(), onceL.end(), func);
std::for_each(onL.begin(), onL.end(), func);
} else {
onceL.clear();
onL.clear();
}
publishing.unlock();
} Connection once(Listener f) {
return onceL.emplace(onceL.cend(), false, std::move(f));
} Connection on(Listener f) {
return onL.emplace(onL.cend(), false, std::move(f));
} void erase(Connection conn) noexcept {
conn->first = true; if(publishing.try_lock()) {
auto pred = [](auto &&element){ return element.first; };
onceL.remove_if(pred);
onL.remove_if(pred);
}
publishing.unlock();
} void run(E event, std::shared_ptr<T> ptr) {
ListenerList currentL;
onceL.swap(currentL); auto func = [&event, &ptr](auto &&element) {
return element.first ? void() : element.second(event, ptr);
}; publishing.lock(); std::for_each(onL.rbegin(), onL.rend(), func);
std::for_each(currentL.rbegin(), currentL.rend(), func); publishing.unlock(); onL.remove_if([](auto &&element){ return element.first; });
} void thread_fun(std::shared_ptr<T> ptr)
{
while(true) {
std::unique_lock<std::mutex> lk(emutex);
econd.wait_for(lk, std::chrono::milliseconds(), [this](){return !events.empty();}); if(events.size() > ) {
E event = std::move(events.front());
events.pop();
run(std::move(event), std::move(ptr));
} else {
break;
}
}
} void publish(E event, std::shared_ptr<T> ptr, bool asyn) {
if(asyn) {
{
std::lock_guard<std::mutex> lk(emutex);
events.push(std::move(event));
}
econd.notify_all(); if(!ethread.joinable()) {
ethread = std::thread(&Handler<E>::thread_fun, this, std::move(ptr));
}
} else {
run(std::move(event), ptr);
}
} void join() noexcept override {
if(ethread.joinable()) {
ethread.join();
}
} void exit() noexcept override {
if(ethread.joinable()) {
econd.notify_all();
ethread.join();
}
} private:
std::mutex publishing;
ListenerList onceL{};
ListenerList onL{}; std::thread ethread;
std::queue<E> events;
std::mutex emutex;
std::condition_variable econd;
};
在里面添加了存放事件的队列，还有用于同步的mutex和条件变量。在publish的时候，区分了异步调用，并且创建了线程。这里写几点重要的东西，
1、创建线程时为了可以传入Emitter，将原来的T&改为了share_ptr<T>，可以看emitter.h源码188～192
template<typename E>
void publish(E event, bool asyn = false) {
// handler<E>().publish(std::move(event), *static_cast<T*>(this), asyn);
handler<E>().publish(std::move(event), this->shared_from_this(), asyn);
}
这里之所以可以用 this->shared_from_this() 来获得类的share_ptr 是因为该类继承了std::enable_shared_from_this，可以看emitter.h源码第33行，类原型大概是这样。
template<typename T>
class Emitter : public std::enable_shared_from_this<T> {}
2、在thread_fun中，设置了默认等待时间为60000ms，也就是1分钟。如果在1分钟内没有事件交由线程处理，那么线程会退出，避免浪费资源。有些看官会说了，那如果我事件就是1分钟发生一次呢，那岂不是每次都要重新创建线程？是的，是需要重新创建，所以大家根据要求来改吧，KeKe～
3、对于相同的事件类型，是否会在线程中处理，和事件的注册没有任何关系，而是在事件发送，也就是publish的时候确定的。考虑到一些场景，比如写日志，同样是日志事件，但是有的日志非常长，需要大量io时间，有的日志比较短，不会浪费很多io时间。那就可以在publish的时候根据日志数据大小来决定，是否需要用异步操作。
4、将原先的bool publishing 改为std::mutex publishing，防止线程中对ListenerList的操作会造成的未知情况。
接下来说一下事件和事件的发送。
对于事件类型，就是一个普通的结构体或类，比如test.cc中的StringEvent 和 IntEvent。但是publish时对Event的构建有时候是可能影响一些性能的，先看test.cc中的48～96：
class A : public Emitter<A>
{
public:
A()
{
cout << "A" << endl;
} void print()
{
publish(StringEvent("Hello"), false);
publish(make_event<StringEvent>("Hello"), true);
publish(make_unique<StringEvent>("World"), true); this_thread::sleep_for(1000ms);
publish(make_unique<IntEvent>(), true);
publish(make_unique<IntEvent>(), true); } ~A()
{
cout << "~A" << endl;
}
}; int main()
{
shared_ptr<A> em = make_shared<A>(); em->on<StringEvent>([](StringEvent& ev, shared_ptr<A>& a){
ev.print();
}); em->on<event_ptr<StringEvent>>([](event_ptr<StringEvent>& ev, shared_ptr<A>& a){
ev->print();
}); em->on<event_ptr<IntEvent>>([](event_ptr<IntEvent>& ev, shared_ptr<A>& a){
ev->print();
}); em->print(); em->thread_join(); return ;
}
把以上的12～17行，注释掉，可以得到结果：
A
StringEvent
string event:Hello
~StringEvent
~StringEvent
~StringEvent
~A
可以看到StringEvent构造了一次，但是却析构了3次，非常恐怖。原因是我们在publish中调用std::move来传递参数。实际上std::move是会调用类的移动构造函数的，但是咱们这里只是在构造函数里打印了一下，所以实际上这里应该是创建了3次StringEvent的，比如这里的StringEvent，移动构造函数的原型应该是
StringEvent(StringEvent&& e){std::cout << "StringEvent" << endl;}
如果把这句加到StringEvent类中去，就会多打印两次 "StringEvent" 了，大家可以动手试试。
很明显，这里使用了移动构造函数，多构建了两次StringEvent，对于Event结构中如果比较复杂，很可能会影响效率，所以我又在emitter.h中加了几行，文件19～26
template<typename E>
using event_ptr = std::unique_ptr<E>; template<typename E, typename... Args>
event_ptr<E> make_event(Args&&... args) {
return std::make_unique<E>(std::forward<Args>(args)...);
}
它的用法在上面代码中有体现，这儿其实就是封装了一下make_unique，用make_unique来构建事件，其实这时候事件类型已经不是E，而是std::unique_ptr<E>，在main函数中有体现，大家可以比对一下看看。这样做的好处就是，Event只构建了一次，某种程度上是会提高点效率的。
好了，这个东西介绍到这里，还没来得及多测试一下，就贴出来了。有问题提出来，大家一起讨论。
下一篇不出意外的话，就继续来看源码，KeKe～
我还不知道博客园哪里可以上传文件，等我研究一下，把代码传上来。再贴链接。 emitter
----------------2017/9/25更新--------------------
上次没来得及仔细测试，下午没事就好好测试了一下，发现里面有很多错误的地方，我已经在代码中进行了修改并重新上传了，下载地址应该还是一样的。
上面写的我就不修改了，作为反面教材吧，KeKe~。另外再说一些东西：
1、thread::joinable，之前我用这个来判断线程是否在运行是错误的，joinable只是返回一种线程状态，用来指明该线程是否可以用join来等待线程结束。如果使用了detch，将线程和主线程分离了，就不能再使用join了。
2、wait_for，之前我把等待条件放在wait_for中的第三个参数，我们调用notify_all后， wait_for会调用匿名函数，如果条件不满足，就继续等直到超时（注意这里的超时时间还是和给定的参数一样）；如果条件满足就返回。现在假设我们wait_for的超时时间非常过长，但是已经没有事件了，这时候我们调用notify_all来终止线程是错误的，原因上面已经说了。这在新代码中作了改进。
3、添加了wait函数，用于等待所有事件的回调结束，并且线程结束。当然这只针对异步的事件。
给大家造成了困扰， 在这里深感抱歉。