UVW源码漫谈（番外篇）—— Emitter

Posted 2020-10-08 yxfangcs

这两天天气凉了，苏州这边连续好几天都是淅淅沥沥的下着小雨，今天天气还稍微好点。前两天早上起来突然就感冒了，当天就用了一卷纸，好在年轻扛得住，第二天就跟没事人似的。在这里提醒大家一下，天气凉了，睡凉席的可以收起来了，体质不太好的，也要适当加点衣服。

本来是想接着看源码的，早上起来又把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.h

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

 

（有大神过来的，可以帮忙看看有没有错误，在评论区留个言，大家可以讨论讨论。）

主要来看看做的一些改动。

先看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～

我还不知道博客园哪里可以上传文件，等我研究一下，把代码传上来。再贴链接。