这两天天气凉了，苏州这边连续好几天都是淅淅沥沥的下着小雨，今天天气还稍微好点。前两天早上起来突然就感冒了，当天就用了一卷纸，好在年轻扛得住，第二天就跟没事人似的。在这里提醒大家一下，天气凉了，睡凉席的可以收起来了，体质不太好的，也要适当加点衣服。
本来是想接着看源码的，早上起来又把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～
我还不知道博客园哪里可以上传文件，等我研究一下，把代码传上来。再贴链接。
时间： 2024-10-09 06:47:03