C++的阻塞式消息总线message_bus实现

一、概述

• 本实现参考自一个通用的C++ 消息总线框架 ,代码位置：messagebus.h
• 该消息总线主要是解决多个模块间的耦合问题，是阻塞式的。
• 消息总线实现的基本原理如下：被通信对象向消息总线发布一个主题，这个主题包含消息主题和消息处理函数，消息主题标示某个特定的主题，消息处理函数用来响应该主题的某种消息类型。通信对象向消息总线发送某个特定主题和消息参数，总线就会根据消息主题和消息参数找到对应的消息处理函数处理该请求。
• 最主要的是这个成员对象boost::unordered_multimap m_map;

二、实现代码

messagebus.h

#pragma once

#include 
#include 
#include 
#include 

#include 
#include 

// 简单的消息分发机制
// ref:https://www.cnblogs.com/qicosmos/archive/2013/04/28/3048919.html
// 《深入应用C++11代码优化与工程级应用》第12章
class MessageBus
{
public:
    // regist message
    template< class... _Args, class _Func, class = typename std::enable_if<!std::is_member_function_pointer<_Func>::value>::type>
    void attach(std::string key, _Func && func)
    {
        std::function<void(_Args...)> fn = [&](_Args... args) { return func(std::forward<_Args>(args)...); };
        m_map.insert(std::make_pair(key, std::move(fn)));
    }

    // non-const member function
    template<class... _Args, class _Class, class... _DeclareArgs, class _Object>
    void attach(std::string key, void(_Class::*func)(_DeclareArgs...), _Object & object)
    {
        std::function<void(_Args...)> fn = [&, func](_Args... args) { return (object.*func)(std::forward<_Args>(args)...); };
        m_map.insert(std::make_pair(key, fn));
    }

    // const member function
    template<class... _Args, class _Class, class... _DeclareArgs, class _Object>
    void attach(std::string key, void(_Class::*func)(_DeclareArgs...) const, _Object & object)
    {
        std::function<void(_Args...)> fn = [&, func](_Args... args) { return (object.*func)(std::forward<_Args>(args)...); };
        m_map.insert(std::make_pair(key, fn));
    }

    // Broadcast messages, call all the callback functions
    template<typename... _Args>
    void sendMessage(std::string key, _Args... args)
    {
        auto range = m_map.equal_range(key);
        for (auto it = range.first; it != range.second; it++)
        {
            std::function<void(_Args...)> func = boost::any_cast<std::function<void(_Args...)>>(it->second);
            func(args...);
        }
    }

    // remove message
    template<typename... _Args>
    void remove(std::string key)
    {
        auto range = m_map.equal_range(key);
        m_map.erase(range.first, range.second);
    }

public:
    MessageBus() = default;
    ~MessageBus() = default;
private:
    MessageBus(const MessageBus&) = delete;
    MessageBus& operator=(const MessageBus&) = delete;

    static std::multimap<std::string, boost::any> m_map;
};

static MessageBus g_messagebus; // 全局消息总线
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71

三、使用方法

• 在任意一cpp文件中实现m_map，比如main函数中；
• 使用全局的g_messagebus在所需订阅模块中订阅感兴趣消息和回调函数；
• 在需要发布消息的代码中填写sendMessage业务逻辑。

#include "messagebus.h"
std::multimap<std::string, boost::any> MessageBus::m_map;

class Test
{
public:
    test(int, string) {}
};
int main()
{
    Test t;
    // 订阅"test"消息，回调函数为Test::test
    g_messagebus.attach<int, string>("test", &t, &Test::test); 
    // 向订阅者发送消息，消息参数是 123 和 "abc"，回调函数Test::test将会被执行
    g_messagebus.sendMessage<int, string>("test", 123, "abc"); 
    // 移除订阅"test"
    g_messagebus.remove<int, string>("test"); 
    return 0;
}
   
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

四、非阻塞消息总线的实现

+请参考我的另一篇博客 : 使用C++11实现的非阻塞消息总线message_bus