ROS发布与订阅处理顺序以及多线程使用

提示：文章写完后，目录可以自动生成，如何生成可参考右边的帮助文档

ROS发布与订阅处理顺序以及多线程使用

• 一、前言
• 二、发布者
• 三、订阅者
• 四、测试结果
  *
• 1、单线程
• 2、多线程
• 3、单独为某个话题开辟线程
• 五、总结

一、前言

记录ROS发布与订阅处理顺序以及多线程使用，由于情况很多，所以不一一细讲，有需要的可以自己尝试。这里简单讲解两个话题以及两个订阅者的情况。

二、发布者

发布方发布两个话题：

talker发布两个话题，然后每个话题都各自开辟自己的队列 （有一些人说之开辟一个队列，个人觉得不严谨，只是因为话题处理是单线程，所以和简化成一个队列类似。这里可以去了解以下话题通讯协议）。如何队列被占满，就会丢弃旧的信息。
可以修改参数：
1、话题1的队列长度
2、话题2的队列长度
3、话题1的发布频率
4、话题2的发布频率

#include "ros/ros.h"
#include "std_msgs/String.h"
#include

int main(int argc, char **argv) {
  ros::init(argc, argv, "talker");
  ros::NodeHandle n;

  ros::Publisher message_a_pub = n.advertise<std_msgs::String>("MessageA", 1000);
  ros::Publisher message_b_pub = n.advertise<std_msgs::String>("MessageB", 1000);

  int message_a_count = 0;
  int message_b_count = 0;

  while (0 == message_a_pub.getNumSubscribers() || 0 == message_b_pub.getNumSubscribers()) {
    ROS_INFO("Waiting for subscribers to connect");
    ros::Duration(0.1).sleep();
  }

  ros::Timer timer_a = n.createTimer(ros::Duration(0.1), [&](const ros::TimerEvent&) {
    std_msgs::String msg;
    std::stringstream ss;
    ss << "Messages A: " << message_a_count;
    msg.data = ss.str();
    ROS_INFO("%s", msg.data.c_str());
    message_a_pub.publish(msg);
    message_a_count++;
  });

  ros::Timer timer_b = n.createTimer(ros::Duration(0.1), [&](const ros::TimerEvent&) {
    std_msgs::String msg;
    std::stringstream ss;
    ss << "Messages B: " << message_b_count;
    msg.data = ss.str();
    ROS_INFO("%s", msg.data.c_str());
    message_b_pub.publish(msg);
    message_b_count++;
  });

  ros::spin();

  return 0;
}

三、订阅者

订阅者订阅两个信息，并在回调函数处理。
由于订阅者是没有所谓的订阅频率的（和程序的处理速度有关），这里使用简单的休眠来模拟”订阅频率”。 简单的话题订阅频率默认和发布频率一样。
可以修改参数：
1、订阅话题1的队列长度
2、订阅话题2的队列长度
3、订阅话题1的发布”频率”（在回调函数设置休眠来模拟）
4、订阅话题2的发布”频率”（在回调函数设置休眠来模拟）

#include
#include
#include
#include "ros/ros.h"
#include "std_msgs/String.h"
using namespace std::chrono_literals;

void CallbackA(const std_msgs::String::ConstPtr& msg) {
  ROS_INFO(" I heard: [%s]", msg->data.c_str());

}

void CallbackB(const std_msgs::String::ConstPtr& msg) {
  ROS_INFO(" I heard: [%s]", msg->data.c_str());
  ros::Rate rate(1);
  rate.sleep();
}

int main(int argc, char **argv) {
  ros::init(argc, argv, "listener");
  ros::NodeHandle n;

  ros::NodeHandle n_a;

  ros::Subscriber sub_a = n_a.subscribe("MessageA", 100, CallbackA);
  ros::Subscriber sub_b = n.subscribe("MessageB", 10, CallbackB);

  ros::MultiThreadedSpinner spinner(2);
  spinner.spin();

  return 0;
}

四、测试结果

1、单线程

1. 测试1

发布者参数设置（1000，1000，0.1，0.1） //分别表示话题1队列长度，话题2队列长度，发布频率，发布频率
订阅者参数设置（1，1，0，0）  //分别表示话题1队列长度，话题2队列长度，订阅频率（0表示不设置休眠），订阅频率（设置休眠0s）

话题交替出现。 如果回调函数是简单数据处理，基本都是按这种理想方式出现，但是如果回调函数处理速度比发布频率慢，那么就对出现各种情况。在这时候，话题的队列长度就可以体现作用。（如果想了解的，可以修改休眠时间和队列长度，去看看实际效果）
2. 测试2
参数：

发布者参数设置（1000，1000，0.1，0.1） //分别表示话题1队列长度，话题2队列长度，发布频率，发布频率
订阅者参数设置（1，1，0，1）  //分别表示话题1队列长度，话题2队列长度，订阅频率（0表示不设置休眠），订阅频率（设置休眠1s）

结果：

订阅话题是按照1Hz频率来订阅，虽然订阅话题1没有设置休眠，但是还是和话题2一样，主要原因是，订阅话题是单线程。就像将订阅话题放进一个队列来处理一样，处理完一个，才能处理下一个。所以导致两个话题发布频率一样。解决方法办法是使用多线程（少量数据的时候都ok，但是长时间处理没去试）。

2、多线程

订阅者代码：

#include
#include
#include
#include "ros/ros.h"
#include "std_msgs/String.h"
using namespace std::chrono_literals;

void CallbackA(const std_msgs::String::ConstPtr& msg) {
  ROS_INFO(" I heard: [%s]", msg->data.c_str());

}

void CallbackB(const std_msgs::String::ConstPtr& msg) {
  ROS_INFO(" I heard: [%s]", msg->data.c_str());
  ros::Rate rate(1);
  rate.sleep();
}

int main(int argc, char **argv) {
  ros::init(argc, argv, "listener");
  ros::NodeHandle n;

  ros::NodeHandle n_a;

  ros::Subscriber sub_a = n_a.subscribe("MessageA", 1, CallbackA);
  ros::Subscriber sub_b = n.subscribe("MessageB", 1, CallbackB);

  ros::MultiThreadedSpinner spinner(2);
  spinner.spin();

  return 0;
}

结果：

话题A的频率变成10Hz，而话题B的频率是1Hz。

3、单独为某个话题开辟线程

#include
#include
#include
#include "ros/ros.h"
#include "std_msgs/String.h"
using namespace std::chrono_literals;

void CallbackA(const std_msgs::String::ConstPtr& msg) {
  ROS_INFO(" I heard: [%s]", msg->data.c_str());

}

void CallbackB(const std_msgs::String::ConstPtr& msg) {
  ROS_INFO(" I heard: [%s]", msg->data.c_str());
  ros::Rate rate(1);
  rate.sleep();
}

int main(int argc, char **argv) {
  ros::init(argc, argv, "listener");
  ros::NodeHandle n;

  ros::NodeHandle n_a;
  ros::CallbackQueue callback_queue_a;
  n_a.setCallbackQueue(&callback_queue_a);
  ros::Subscriber sub_a = n_a.subscribe("MessageA", 1, CallbackA);
  ros::Subscriber sub_b = n.subscribe("MessageB", 1, CallbackB);

  std::thread spinner_thread_a([&callback_queue_a]() {
    ros::SingleThreadedSpinner spinner_a;
    spinner_a.spin(&callback_queue_a);
  });

  ros::spin();
  spinner_thread_a.join();

  return 0;
}

结果：

和多线程一样。

五、总结

1. 话题发布基本流程：每一个话题都各自开辟自己队列，然后把数据发布到队列中，等待订阅者读取，如果订阅者来不及处理，那么当发布者队列被沾满后，就会将旧的数据覆盖。
2. 订阅者的订阅频率取决于回调函数的处理速度，单线程时，回调函数的频率是一样的，等于所有回调函数处理时间之和的倒数。
3. 如果多线程，可以避免部分数据丢失。

注：实际代码可能会出现更复杂情况，而不是简单的休眠来替代回调函数处理时间。

Original: https://blog.csdn.net/qq_16539009/article/details/123526165
Author: 柯藤
Title: ROS发布与订阅处理顺序以及多线程使用