这两天天气凉了，苏州这边连续好几天都是淅淅沥沥的下着小雨，今天天气还稍微好点。前两天早上起来突然就感冒了，当天就用了一卷纸，好在年轻扛得住，第二天就跟没事人似的。在这里提醒大家一下，天气凉了，睡凉席的可以收起来了，体质不太好的，也要适当加点衣服。
本来是想接着看源码的，早上起来又把Emitter鼓捣了一下，跟大家说说。
emitter.hpp是可以从源码中剥离出来的，只要去除里面的libuv的东西就行了。Emitter其实就是实现的即时回调，没有异步事件处理的功能。但是我们有时候是需要用并发来提高处理速度的，于是我就把Emitter稍微改造了一下，先上代码：
1 #pragma once 2 3 4 #include <type_traits> 5 #include <functional> 6 #include <algorithm> 7 #include <utility> 8 #include <cstddef> 9 #include <vector> 10 #include <memory> 11 #include <list> 12 #include <queue> 13 #include <thread> 14 #include <mutex> 15 #include <condition_variable> 16 #include <chrono> 17 18 19 template<typename E> 20 using event_ptr = std::unique_ptr<E>; 21 22 23 template<typename E, typename... Args> 24 event_ptr<E> make_event(Args&&... args) { 25 return std::make_unique<E>(std::forward<Args>(args)...); 26 } 27 28 29 /** 30 * @brief Event emitter base class. 31 */ 32 template<typename T> 33 class Emitter : public std::enable_shared_from_this<T> { 34 struct BaseHandler { 35 virtual ~BaseHandler() noexcept = default; 36 virtual bool empty() const noexcept = 0; 37 virtual void clear() noexcept = 0; 38 virtual void join() noexcept = 0; 39 virtual void exit() noexcept = 0; 40 }; 41 42 template<typename E> 43 struct Handler final: BaseHandler { 44 using Listener = std::function<void(E &, std::shared_ptr<T>&)>; 45 using Element = std::pair<bool, Listener>; 46 using ListenerList = std::list<Element>; 47 using Connection = typename ListenerList::iterator; 48 49 bool empty() const noexcept override { 50 auto pred = [](auto &&element){ return element.first; }; 51 52 return std::all_of(onceL.cbegin(), onceL.cend(), pred) && 53 std::all_of(onL.cbegin(), onL.cend(), pred); 54 } 55 56 void clear() noexcept override { 57 if(!publishing.try_lock()) { 58 auto func = [](auto &&element){ element.first = true; }; 59 std::for_each(onceL.begin(), onceL.end(), func); 60 std::for_each(onL.begin(), onL.end(), func); 61 } else { 62 onceL.clear(); 63 onL.clear(); 64 } 65 publishing.unlock(); 66 } 67 68 Connection once(Listener f) { 69 return onceL.emplace(onceL.cend(), false, std::move(f)); 70 } 71 72 Connection on(Listener f) { 73 return onL.emplace(onL.cend(), false, std::move(f)); 74 } 75 76 void erase(Connection conn) noexcept { 77 conn->first = true; 78 79 if(publishing.try_lock()) { 80 auto pred = [](auto &&element){ return element.first; }; 81 onceL.remove_if(pred); 82 onL.remove_if(pred); 83 } 84 publishing.unlock(); 85 } 86 87 void run(E event, std::shared_ptr<T> ptr) { 88 ListenerList currentL; 89 onceL.swap(currentL); 90 91 auto func = [&event, &ptr](auto &&element) { 92 return element.first ? void() : element.second(event, ptr); 93 }; 94 95 publishing.lock(); 96 97 std::for_each(onL.rbegin(), onL.rend(), func); 98 std::for_each(currentL.rbegin(), currentL.rend(), func); 99 100 publishing.unlock(); 101 102 onL.remove_if([](auto &&element){ return element.first; }); 103 } 104 105 void thread_fun(std::shared_ptr<T> ptr) 106 { 107 while(true) { 108 std::unique_lock<std::mutex> lk(emutex); 109 econd.wait_for(lk, std::chrono::milliseconds(60000), [this](){return !events.empty();}); 110 111 if(events.size() > 0) { 112 E event = std::move(events.front()); 113 events.pop(); 114 run(std::move(event), std::move(ptr)); 115 } else { 116 break; 117 } 118 } 119 } 120 121 void publish(E event, std::shared_ptr<T> ptr, bool asyn) { 122 if(asyn) { 123 { 124 std::lock_guard<std::mutex> lk(emutex); 125 events.push(std::move(event)); 126 } 127 econd.notify_all(); 128 129 if(!ethread.joinable()) { 130 ethread = std::thread(&Handler<E>::thread_fun, this, std::move(ptr)); 131 } 132 } else { 133 run(std::move(event), ptr); 134 } 135 } 136 137 void join() noexcept override { 138 if(ethread.joinable()) { 139 ethread.join(); 140 } 141 } 142 143 void exit() noexcept override { 144 if(ethread.joinable()) { 145 econd.notify_all(); 146 ethread.join(); 147 } 148 } 149 150 private: 151 std::mutex publishing; 152 ListenerList onceL{}; 153 ListenerList onL{}; 154 155 std::thread ethread; 156 std::queue<E> events; 157 std::mutex emutex; 158 std::condition_variable econd; 159 }; 160 161 static std::size_t next_type() noexcept { 162 static std::size_t counter = 0; 163 return counter++; 164 } 165 166 template<typename> 167 static std::size_t event_type() noexcept { 168 static std::size_t value = next_type(); 169 return value; 170 } 171 172 template<typename E> 173 Handler<E> & handler() noexcept { 174 std::size_t type = event_type<E>(); 175 176 if(!(type < handlers.size())) { 177 handlers.resize(type+1); 178 } 179 180 if(!handlers[type]) { 181 handlers[type] = std::make_unique<Handler<E>>(); 182 } 183 184 return static_cast<Handler<E>&>(*handlers[type]); 185 } 186 187 protected: 188 template<typename E> 189 void publish(E event, bool asyn = false) { 190 // handler<E>().publish(std::move(event), *static_cast<T*>(this), asyn); 191 handler<E>().publish(std::move(event), this->shared_from_this(), asyn); 192 } 193 194 public: 195 template<typename E> 196 using Listener = typename Handler<E>::Listener; 197 198 /** 199 * @brief Connection type for a given event type. 200 * 201 * Given an event type `E`, `Connection<E>` is the type of the connection 202 * object returned by the event emitter whenever a listener for the given 203 * type is registered. 204 */ 205 template<typename E> 206 struct Connection: private Handler<E>::Connection { 207 template<typename> friend class Emitter; 208 209 Connection() = default; 210 Connection(const Connection &) = default; 211 Connection(Connection &&) = default; 212 213 Connection(typename Handler<E>::Connection conn) 214 : Handler<E>::Connection{std::move(conn)} 215 {} 216 217 Connection & operator=(const Connection &) = default; 218 Connection & operator=(Connection &&) = default; 219 }; 220 221 virtual ~Emitter() noexcept { 222 static_assert(std::is_base_of<Emitter<T>, T>::value, "!"); 223 } 224 225 /** 226 * @brief Registers a long-lived listener with the event emitter. 227 * 228 * This method can be used to register a listener that is meant to be 229 * invoked more than once for the given event type.<br/> 230 * The Connection object returned by the method can be freely discarded. It 231 * can be used later to disconnect the listener, if needed. 232 * 233 * Listener is usually defined as a callable object assignable to a 234 * `std::function<void(const E &, T &)`, where `E` is the type of the event 235 * and `T` is the type of the resource. 236 * 237 * @param f A valid listener to be registered. 238 * @return Connection object to be used later to disconnect the listener. 239 */ 240 template<typename E> 241 Connection<E> on(Listener<E> f) { 242 return handler<E>().on(std::move(f)); 243 } 244 245 /** 246 * @brief Registers a short-lived listener with the event emitter. 247 * 248 * This method can be used to register a listener that is meant to be 249 * invoked only once for the given event type.<br/> 250 * The Connection object returned by the method can be freely discarded. It 251 * can be used later to disconnect the listener, if needed. 252 * 253 * Listener is usually defined as a callable object assignable to a 254 * `std::function<void(const E &, T &)`, where `E` is the type of the event 255 * and `T` is the type of the resource. 256 * 257 * @param f Avalid listener to be registered. 258 * @return Connection object to be used later to disconnect the listener. 259 */ 260 template<typename E> 261 Connection<E> once(Listener<E> f) { 262 return handler<E>().once(std::move(f)); 263 } 264 265 /** 266 * @brief Disconnects a listener from the event emitter. 267 * @param conn A valid Connection object 268 */ 269 template<typename E> 270 void erase(Connection<E> conn) noexcept { 271 handler<E>().erase(std::move(conn)); 272 } 273 274 /** 275 * @brief Disconnects all the listeners for the given event type. 276 */ 277 template<typename E> 278 void clear() noexcept { 279 handler<E>().clear(); 280 } 281 282 /** 283 * @brief Disconnects all the listeners. 284 */ 285 void clear() noexcept { 286 std::for_each(handlers.begin(), handlers.end(), 287 [](auto &&hdlr){ if(hdlr) { hdlr->clear(); } }); 288 } 289 290 /** 291 * @brief Checks if there are listeners registered for the specific event. 292 * @return True if there are no listeners registered for the specific event, 293 * false otherwise. 294 */ 295 template<typename E> 296 bool empty() const noexcept { 297 std::size_t type = event_type<E>(); 298 299 return (!(type < handlers.size()) || 300 !handlers[type] || 301 static_cast<Handler<E>&>(*handlers[type]).empty()); 302 } 303 304 /** 305 * @brief Checks if there are listeners registered with the event emitter. 306 * @return True if there are no listeners registered with the event emitter, 307 * false otherwise. 308 */ 309 bool empty() const noexcept { 310 return std::all_of(handlers.cbegin(), handlers.cend(), 311 [](auto &&hdlr){ return !hdlr || hdlr->empty(); }); 312 } 313 314 void thread_join() const noexcept { 315 std::for_each(handlers.begin(), handlers.end(), 316 [](auto &&hdlr){ if(hdlr) { hdlr->join(); } }); 317 } 318 319 void thread_exit() const noexcept { 320 std::for_each(handlers.begin(), handlers.end(), 321 [](auto &&hdlr){ if(hdlr) { hdlr->exit(); } }); 322 } 323 324 private: 325 std::vector<std::unique_ptr<BaseHandler>> handlers{}; 326 };
Emitter类应该和项目是兼容的，但是为了更干净和通用一点，去除了ErrorEvent这些东西，所以已经不适合再放到源码里。下面是使用的例子：
1 #include <iostream> 2 #include <memory> 3 #include <thread> 4 5 using namespace std; 6 #include "emitter.h" 7 8 9 struct StringEvent 10 { 11 StringEvent(std::string str):i_str(str) 12 { 13 cout << "StringEvent" << std::endl; 14 } 15 16 void print() 17 { 18 std::cout << "string event:" << i_str << std::endl; 19 } 20 21 std::string i_str; 22 23 ~StringEvent() 24 { 25 cout << "~StringEvent" << std::endl; 26 } 27 }; 28 29 struct IntEvent 30 { 31 IntEvent(int t) : i_t(t) 32 { 33 cout << "IntEvent" << std::endl; 34 } 35 void print() 36 { 37 std::cout << "int event:" << i_t << std::endl; 38 } 39 ~IntEvent() 40 { 41 cout << "~IntEvent" << std::endl; 42 } 43 44 int i_t{1}; 45 }; 46 47 48 class A : public Emitter<A> 49 { 50 public: 51 A() 52 { 53 cout << "A" << endl; 54 } 55 56 void print() 57 { 58 publish(StringEvent("Hello"), false); 59 publish(make_event<StringEvent>("Hello"), true); 60 publish(make_unique<StringEvent>("World"), true); 61 62 this_thread::sleep_for(1000ms); 63 publish(make_unique<IntEvent>(2), true); 64 publish(make_unique<IntEvent>(3), true); 65 66 } 67 68 ~A() 69 { 70 cout << "~A" << endl; 71 } 72 }; 73 74 75 int main() 76 { 77 shared_ptr<A> em = make_shared<A>(); 78 79 em->on<StringEvent>([](StringEvent& ev, shared_ptr<A>& a){ 80 ev.print(); 81 }); 82 83 em->on<event_ptr<StringEvent>>([](event_ptr<StringEvent>& ev, shared_ptr<A>& a){ 84 ev->print(); 85 }); 86 87 em->on<event_ptr<IntEvent>>([](event_ptr<IntEvent>& ev, shared_ptr<A>& a){ 88 ev->print(); 89 }); 90 91 em->print(); 92 93 em->thread_join(); 94 95 return 0; 96 }
主要来看看做的一些改动。
先看test.cc里面，现在的事件处理函数Lambda中的两个参数做了改变：
1 em->on<StringEvent>([](StringEvent& ev, shared_ptr<A>& a){ 2 ev.print(); 3 });
第二个参数由原来的 A& 类型 改成了 share_ptr<A>& 类型。
1 class A : public Emitter<A> 2 { 3 public: 4 A() 5 { 6 cout << "A" << endl; 7 } 8 9 void print() 10 { 11 publish(StringEvent("Hello"), false); 12 publish(make_event<StringEvent>("Hello"), true); 13 publish(make_unique<StringEvent>("World"), true); 14 15 this_thread::sleep_for(1000ms); 16 publish(make_unique<IntEvent>(2), true); 17 publish(make_unique<IntEvent>(3), true); 18 19 } 20 21 ~A() 22 { 23 cout << "~A" << endl; 24 } 25 };
我们看过之前的代码，现在在publish中多加了一个bool参数，默认值为false，用于指示这个事件是否需要异步处理。
这里注意下，指示异步处理的是在发生事件，调用publish的时候。
用法上面主要就这些改动，然后再来看emitter.h
Handler类
1 template<typename E> 2 struct Handler final: BaseHandler { 3 using Listener = std::function<void(E &, std::shared_ptr<T>&)>; 4 using Element = std::pair<bool, Listener>; 5 using ListenerList = std::list<Element>; 6 using Connection = typename ListenerList::iterator; 7 8 bool empty() const noexcept override { 9 auto pred = [](auto &&element){ return element.first; }; 10 11 return std::all_of(onceL.cbegin(), onceL.cend(), pred) && 12 std::all_of(onL.cbegin(), onL.cend(), pred); 13 } 14 15 void clear() noexcept override { 16 if(!publishing.try_lock()) { 17 auto func = [](auto &&element){ element.first = true; }; 18 std::for_each(onceL.begin(), onceL.end(), func); 19 std::for_each(onL.begin(), onL.end(), func); 20 } else { 21 onceL.clear(); 22 onL.clear(); 23 } 24 publishing.unlock(); 25 } 26 27 Connection once(Listener f) { 28 return onceL.emplace(onceL.cend(), false, std::move(f)); 29 } 30 31 Connection on(Listener f) { 32 return onL.emplace(onL.cend(), false, std::move(f)); 33 } 34 35 void erase(Connection conn) noexcept { 36 conn->first = true; 37 38 if(publishing.try_lock()) { 39 auto pred = [](auto &&element){ return element.first; }; 40 onceL.remove_if(pred); 41 onL.remove_if(pred); 42 } 43 publishing.unlock(); 44 } 45 46 void run(E event, std::shared_ptr<T> ptr) { 47 ListenerList currentL; 48 onceL.swap(currentL); 49 50 auto func = [&event, &ptr](auto &&element) { 51 return element.first ? void() : element.second(event, ptr); 52 }; 53 54 publishing.lock(); 55 56 std::for_each(onL.rbegin(), onL.rend(), func); 57 std::for_each(currentL.rbegin(), currentL.rend(), func); 58 59 publishing.unlock(); 60 61 onL.remove_if([](auto &&element){ return element.first; }); 62 } 63 64 void thread_fun(std::shared_ptr<T> ptr) 65 { 66 while(true) { 67 std::unique_lock<std::mutex> lk(emutex); 68 econd.wait_for(lk, std::chrono::milliseconds(60000), [this](){return !events.empty();}); 69 70 if(events.size() > 0) { 71 E event = std::move(events.front()); 72 events.pop(); 73 run(std::move(event), std::move(ptr)); 74 } else { 75 break; 76 } 77 } 78 } 79 80 void publish(E event, std::shared_ptr<T> ptr, bool asyn) { 81 if(asyn) { 82 { 83 std::lock_guard<std::mutex> lk(emutex); 84 events.push(std::move(event)); 85 } 86 econd.notify_all(); 87 88 if(!ethread.joinable()) { 89 ethread = std::thread(&Handler<E>::thread_fun, this, std::move(ptr)); 90 } 91 } else { 92 run(std::move(event), ptr); 93 } 94 } 95 96 void join() noexcept override { 97 if(ethread.joinable()) { 98 ethread.join(); 99 } 100 } 101 102 void exit() noexcept override { 103 if(ethread.joinable()) { 104 econd.notify_all(); 105 ethread.join(); 106 } 107 } 108 109 private: 110 std::mutex publishing; 111 ListenerList onceL{}; 112 ListenerList onL{}; 113 114 std::thread ethread; 115 std::queue<E> events; 116 std::mutex emutex; 117 std::condition_variable econd; 118 };
在里面添加了存放事件的队列，还有用于同步的mutex和条件变量。在publish的时候，区分了异步调用，并且创建了线程。这里写几点重要的东西，
1、创建线程时为了可以传入Emitter，将原来的T&改为了share_ptr<T>，可以看emitter.h源码188～192
1 template<typename E> 2 void publish(E event, bool asyn = false) { 3 // handler<E>().publish(std::move(event), *static_cast<T*>(this), asyn); 4 handler<E>().publish(std::move(event), this->shared_from_this(), asyn); 5 }
这里之所以可以用 this->shared_from_this() 来获得类的share_ptr 是因为该类继承了std::enable_shared_from_this，可以看emitter.h源码第33行，类原型大概是这样。
1 template<typename T> 2 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：
1 class A : public Emitter<A> 2 { 3 public: 4 A() 5 { 6 cout << "A" << endl; 7 } 8 9 void print() 10 { 11 publish(StringEvent("Hello"), false); 12 publish(make_event<StringEvent>("Hello"), true); 13 publish(make_unique<StringEvent>("World"), true); 14 15 this_thread::sleep_for(1000ms); 16 publish(make_unique<IntEvent>(2), true); 17 publish(make_unique<IntEvent>(3), true); 18 19 } 20 21 ~A() 22 { 23 cout << "~A" << endl; 24 } 25 }; 26 27 28 int main() 29 { 30 shared_ptr<A> em = make_shared<A>(); 31 32 em->on<StringEvent>([](StringEvent& ev, shared_ptr<A>& a){ 33 ev.print(); 34 }); 35 36 em->on<event_ptr<StringEvent>>([](event_ptr<StringEvent>& ev, shared_ptr<A>& a){ 37 ev->print(); 38 }); 39 40 em->on<event_ptr<IntEvent>>([](event_ptr<IntEvent>& ev, shared_ptr<A>& a){ 41 ev->print(); 42 }); 43 44 em->print(); 45 46 em->thread_join(); 47 48 return 0; 49 }
把以上的12～17行，注释掉，可以得到结果：
1 A 2 StringEvent 3 string event:Hello 4 ~StringEvent 5 ~StringEvent 6 ~StringEvent 7 ~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
1 template<typename E> 2 using event_ptr = std::unique_ptr<E>; 3 4 5 template<typename E, typename... Args> 6 event_ptr<E> make_event(Args&&... args) { 7 return std::make_unique<E>(std::forward<Args>(args)...); 8 }
它的用法在上面代码中有体现，这儿其实就是封装了一下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函数，用于等待所有事件的回调结束，并且线程结束。当然这只针对异步的事件。
给大家造成了困扰， 在这里深感抱歉。
内容来源于网络如有侵权请私信删除
- 还没有人评论，欢迎说说您的想法！
客服
