转自：https://blog.csdn.net/u012423865/article/details/80024870
How to advertise and subscribe a topic
导读
众所周知，Apollo是基于ROS开发的，所以其底层也是基于消息的机制进行节点通信的。但是它在ROS的基础上做了一些改动，如下：
P2P——由于原生ROS的消息机制是通过主节点(Master)分发数据来实现的，这样一个强中心化的结构始终存在一个Master意外导致系统奔溃的隐患，为了解决这个问题，Apollo使用了Fast-RTPS，用P2P的方式抛弃了主节点实现通信。
Protobuf——原生支持Google的Protobuf，完美的解决了原来ROS的MD5验证导致消息不能后向兼容的问题。
共享内存——使用共享内存传输，效率更高。
上面简单的介绍了一些Apollo和ROS的渊源，如果想详细了解可以看Apollo-ROS官方详细文档。鉴于与ROS的关系，那么就不得不说如何在Apollo里面利用topic进行通信了。虽然出自ROS，但是Apollo在ROS的基础上进行了大量的封装和改动，所以，较原生的ROS，在advertise/subscribe（通知/订阅）topic有了较大的区别。
本文我将按照操作步骤、代码分析、流程图的步骤简述一下Apollo的topic使用方法，如果你只想知道如何使用，那么看前面的使用方法即可，如果你想刨根问题，那么请耐心的看完，结合上下文一起观看，效果更好。还有，代码的注释不能放过，有一些上下文联系的地方，我可能放在代码注释里面来解释了。
使用方法
订阅
订阅已配置好的Topic
实现回调函数
void ZuoTestSubnode::ImgTestCallback(const sensor_msgs::Image &msg){
AINFO << "ImgTestCallback";
//-- do sth
}
1
2
3
4
在InitInternal()内将回调函数添加到对应Topic的回调函数队列，如下：
bool ZuoTestSubnode::InitInternal(){
AdapterManager::AddImageShortCallback(&ZuoTestSubnode::ImgTestCallback, this);
}
1
2
3
订阅未配置好的Topic
如果用户需要新增Topic，那就需要重新配置，而配置，本质上就是调用subscribe/advertise函数，然后生成空的回调函数队列和消息发布句柄。这些是Apollo对ROS做的封装，主要目的是为了方便使用，以及管理众多的topic，这部分主要是在adapter_manager.h、message_manager.h等实现的。具体代码后面再讲，先看步骤：
在adapter_gflags.cc和adapter_gflags.h里面添加新topic。如下：
//-- adapter_gflags.h
//-- Zuo added on 2018-04-15 for testPublishSubnode
DECLARE_string(zuo_test_topic);
1
2
3
4
//-- adapter_gflags.cc
//-- Zuo added on 2018-04-15 for testPublishSubnode
DEFINE_string(zuo_test_topic, "/apollo/zuo/zuo_test", "Zuo added for testPublishMsg");
1
2
3
4
在adapter_manager.h里面增加对ZuoTest的适配：
class AdapterManager {
public:
......
//-- Zuo added on 2018-04-15 for testPublishSubnode
//-- 这里的ZuoTest可以理解为对topic取了个别名。
REGISTER_ADAPTER(ZuoTest);
};
1
2
3
4
5
6
7
在adapter_manager.cc的**Init()**函数的有限状态机内添加一个分支，如下：
void AdapterManager::Init(const AdapterManagerConfig &configs) {
......
for (const auto &config : configs.config()) {
switch (config.type()) {
......
case AdapterConfig::ZUO_TEST:
EnableZuoTest(FLAGS_zuo_test_topic, config);
break;
default:
AERROR << "Unknown adapter config type!";
break;
}
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
在adapter_config.proto里面增加ZUO_TEST标签：（不然上述的有限状态机分支就无法进入）
message AdapterConfig {
enum MessageType {
......
ZUO_TEST = 45;
}
1
2
3
4
5
在message_adapters.h里面增加对应消息的别名：
using ZuoTestAdapter = Adapter<sensor_msgs::Image>;
1
剩下的就和上述已配置好的Topic的使用方法相同了，我就不赘述了。
在使用者的adapter.conf文件内添加一个config标签，这个conf文件是用来配置对应的topic。而config中的type，是用来在后文的有限状态机中，进入我们新添的topic分支。这里的adapter.conf是使用者用来配置对topic的subscribe和advertise的。
config {
type: ZUO_TEST
mode: PUBLISH_ONLY
message_history_limit: 5
}
1
2
3
4
5
所谓的使用者，可以理解为某个模块，例如Perception模块里面就有用来配置这个的。这必定是每个模块的第一个操作，先subscribe和advertise之后才能操作topic咯。
Status Perception::Init() {
AdapterManager::Init(FLAGS_perception_adapter_config_filename);
......
}
1
2
3
4
通知
到这里就肯定都已经配置好topic了，所以都一样了就是调用Publish函数发布数据到topic。
//-- 新建msg
PerceptionObstacles obstacles;
//-- publish msg
common::adapter::AdapterManager::PublishPerceptionObstacles(obstacles);
```
1
2
3
4
5
代码分析
上一篇How_to_add_a_subnode已经讲到了如何添加一个subnode，但是还没有说到如何在这个subnode里面subscribe/publish(订阅/发布)消息，下面将根据上述的操作步骤讲解在Apollo里面为什么要这么用topic。
在说Apollo前，还是带一下ROS的通信机制，也方便我们后续的理解，详情可以了解ROS Wiki
原生的ROS其实是如上图一个消息传递流程，简而言之就是，一个节点发，一个节点收，中间是由Master进行转发，具体操作如下：
//-- Zuo added on 2018-04-13
void chatterCallback(const std_msgs::String::ConstPtr& msg)
{
//-- do sth
printf("msg's data = %s", msg.data);
}
int main(int argc, char **argv)
{
ros::init(argc, argv, "listener");
ros::NodeHandle n;
//-- 向Topic发布消息
//-- 这里是告诉Master，我需要在'TopicfullName'上发布一个消息。
//-- @param_0 Topic名字
//-- @param_1 消息发布队列大小
//-- 返回的句柄需要保存，用来调用publish函数
ros::Publisher Zuo_pub = n.advertise<std_msgs::String>("TopicfullName", 1000);
std_msgs::String msg;
msg.data = "Hello World";
//-- 用户根据返回的句柄调用publish函数，向对应的topic发送消息
Zuo_pub.publish(msg);
//-- 订阅一个Topic
//-- 如果`TopicfullName`接收到消息，就会触发这里的回调函数chatterCallback()
ros::Subscriber sub = n.subscribe("TopicfullName", 1000, chatterCallback);
ros::spin();//ros::spin()进入自循环，循环获取事件
return 0;
}
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
注意：上面的两个函数advertise/subscribe的方式我们可以多留意一下，之后在Apollo的代码讲解里面也会看到类似的方式。这个在Apollo里面都是放在AdapterManager::Init里面的EnableXXX()来调用了，注意看adapter_manager.h的内部实现。
在Apollo里面，为了降低subnode在通信过程中对Master节点的依赖，将上图的机制设计成两个subnode直接类似P2P通信，相当于对原来的网络做了一个去中心化的改变。但是这里我们不用管，对于用户来说，我们的使用方式没变，还是通过调用advertise/subscribe两个接口来进行节点通信。
诚然，在Apollo里面我们可以直接调用advertise/subscribe，但是可以用，不代表应该用。看代码我们会发现，在Apollo底层（adapter_manager.h）的确是调用了advertise/subscribe，但是在其之上还封装了几层，它将对不同Topic的advertise/subscribe使用封装成了对应的带有Topic名字标签的函数，例如：
common::adapter::AdapterManager::PublishPerceptionObstacles(obstacles);
1
上述这种本质上就是ROS里面的publish()函数，但是Apollo对publish()函数封装后，使得每个不同模块的publish都带有各自不同的标签，每一个PublishXXX()只能发送对应类型的Msg，也就是说，原生的ROS下，在大量的发送/订阅的场景下，需要用户管理大量的句柄（这里可以参考上述ROS的简例）。
上面是Publish，下面看看Apollo对subscribe是怎么封装的：
AdapterManager::AddImageShortCallback(&ZuoTestSubnode::ZuoImgTestCallback,this);
1
Apollo将ROS的回调函数封装成一个函数队列，上面的函数就是往这个函数队列里添加回调函数。那么，这个函数在哪里定义的呢？如果直接搜索，是找不到的。这里Apollo用了一个技巧。AdapterManager::AddImageShortCallback()是在adapter_manager里面用宏展开和##拼接结合的方法而成。通过调用REGISTER_ADAPTER(name)来实现（REGISTER_ADAPTER的调用参考操作步骤里面的适配操作）。可以看到adapter_manager.h:#90开始，定义了三个不同参数的static void Add##name##Callback()函数。如下：
static void Add##name##Callback(name##Adapter::Callback callback) { \
CHECK(instance()->name##_) \
<< "Initialize adapter before setting callback"; \
instance()->name##_->AddCallback(callback); \
} \
template <class T> \
static void Add##name##Callback( \
void (T::*fp)(const name##Adapter::DataType &data), T *obj) { \
Add##name##Callback(std::bind(fp, obj, std::placeholders::_1)); \
} \
template <class T> \
static void Add##name##Callback( \
void (T::*fp)(const name##Adapter::DataType &data)) { \
Add##name##Callback(fp); \
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
我们先不管函数具体作用，但是从上述代码我们可以看到，后两个函数最终也是调用了第一个函数，也就是
static void Add##name##Callback(name##Adapter::Callback callback) { \
1
接着跟进去，我们看到这个函数实际上就是调用了：
instance()->name##_->AddCallback(callback); \
1
再跟进，可以看到在adapter.h里：
/**
* @brief registers the provided callback function to the adapter,
* so that the callback function will be called once right after the
* message hits the adapter.
* @param callback the callback with signature void(const D &).
*/
void AddCallback(Callback callback) {
receive_callbacks_.push_back(callback);
}
1
2
3
4
5
6
7
8
9
receive_callbacks_的定义如下：
/// User defined function when receiving a message
std::vector<Callback> receive_callbacks_;
1
2
其实到这里，Apollo订阅的封装实现很明显了，其在adapter_manager.h通过宏展开和拼接生成带有不同名字标签的函数群，这些函数群包含一系列topic相关操作的功能，如订阅，压入回调函数队列等等，然后在adapter_manager.cc里面通过有限状态机根据读入的config文件（参考dag_config_path变量）配置需要启动的subnode。这里调用的Enable##name(topicName，config)实现了对不同topic的订阅以及生成通知句柄，然后就可以调用Add##name##Callback来订阅指定的topic，以及调用Publish##name向指定topic发送消息。
那么Apollo这样绕了一圈将advertise/subscribe封装起来，有什么用呢？原因有如下两点：
当系统里面的相似类型多了，变量的管理会变得复杂起来，那么用这种带有名字标签的命名方式，可以让变量名能够做到顾名思义。
本质上，不同的subscribe和publish函数都是一些相同代码，Apollo通过##拼接和宏展开结合的方式，提高了这一块的代码复用率，也方便整个框架后续的延伸扩展。
还有一个要注意，name##Adapter是不同的消息格式，在message_adapters.h定义的别名，这是为了配合整套方案而将不同的消息都别名成了nameAdapter的形式。
using PerceptionObstaclesAdapter = Adapter<perception::PerceptionObstacles>;
1
类似上面的这样取别名，这里的perception::PerceptionObstacles是在perception_obstacle.pb.h的文件内定义的了，这个和protobuf有关，暂且不说，这是下一章How_to_add_a_new_msg的内容。
流程图
下面是整个topic配置的整体流程图，最终呈现在用户面前的是两句Publish##Name()和Add##Name##Callbackc()。
---------------------
作者：Zuo丶
来源：CSDN
原文：https://blog.csdn.net/u012423865/article/details/80024870
版权声明：本文为博主原创文章，转载请附上博文链接！
