UVW源碼漫談（番外篇）—— Emitter
這兩天天氣涼了，蘇州這邊連續好幾天都是淅淅瀝瀝的下著小雨，今天天氣還稍微好點。前兩天早上起來突然就感冒了，當天就用了一卷紙，好在年輕扛得住，第二天就跟沒事人似的。在這里提醒大家一下，天氣涼了，睡涼席的可以收起來了，體質不太好的，也要適當加點衣服。
本來是想接著看源碼的，早上起來又把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函數，用于等待所有事件的回調結束，并且線程結束。當然這只針對異步的事件。
給大家造成了困擾， 在這里深感抱歉。
浙公網安備 33010602011771號