ROS AsyncSpinner 的多线程行为

我试图了解 ROS 中的 AsyncSpinner 是如何工作的，因为我可能有一些误解。你可以找到类似的问题here.

As seen here它的定义提到：

异步旋转器：产生几个线程（可配置） 将并行执行回调，同时不会阻塞执行该操作的线程 叫它。启动/停止方法允许控制回调的时间 开始处理以及何时应停止。

并且在官方文档中hereAsyncSpinning也被称为多线程Spinning的一种。

因此，我有一个非常简单的示例，其中发布者和订阅者使用 AsyncSpinner 来测试多线程行为。

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

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

  ros::Publisher chatter_pub = nh.advertise<std_msgs::String>("chatter", 1000);

  ros::Rate loop_rate(10);
  while (ros::ok())
  {
    std_msgs::String msg;
    msg.data = "hello world";

    chatter_pub.publish(msg);

    ros::spinOnce();

    loop_rate.sleep();
  }

  return 0;
}

以及定义和使用微调器的订阅者：

#include "ros/ros.h"
#include "std_msgs/String.h"
#include <boost/thread.hpp>

int count = 0;

void chatterCallback(const std_msgs::String::ConstPtr& msg)
{
  count++;
  ROS_INFO("Subscriber %i callback: I heard %s", count, msg->data.c_str());
  sleep(1);
}

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

  ros::Subscriber sub = nh.subscribe("chatter", 1000, chatterCallback);

  ros::AsyncSpinner spinner(boost::thread::hardware_concurrency());
  ros::Rate r(10);

  spinner.start();
  ros::waitForShutdown();

  return 0;
}

当我运行这两个程序时，我得到以下输出：

[ INFO] [1517215527.481856914]: Subscriber 1 callback: I heard hello world
[ INFO] [1517215528.482005146]: Subscriber 2 callback: I heard hello world
[ INFO] [1517215529.482204798]: Subscriber 3 callback: I heard hello world

正如您所看到的，回调每秒运行一次，并且没有并行调用其他回调。我知道全局回调队列正在履行，因为如果我停止发布者，订阅者将继续从队列中弹出累积的消息。

我知道我不应该阻止回调，但在上面的定义中指出，这不会停止调用它的线程，我猜旋转器创建的其他线程也不会停止。我是否仅仅因为阻止回调而阻止下一个回调？难道我有什么误解？我有点困惑，无法证明回调是并行运行的。也许你还有另一个例子？

简短回答：

ROS 回调默认是线程安全的。这意味着注册的回调只能由一个线程处理，并且并发调用被禁用。第二个线程无法同时访问同一回调。

如果您注册第二个回调，您将看到微调器按预期工作，并且多个线程同时调用您的回调。

ros::Subscriber sub1 = nh.subscribe("chatter", 1000, chatterCallback);
ros::Subscriber sub2 = nh.subscribe("chatter", 1000, chatterCallback);

扩展答案：

异步旋转器尝试call回调队列中的可用回调以速率允许的速度进行。如果回调已经在处理中（由另一个线程）CallResult is TryAgain。这意味着稍后将开始新的尝试。

The 这个锁的实现使用变量allow_concurrent_callbacks_这意味着此行为是可选的。

解决方案：

通过设置正确的值可以允许并发调用SubscribeOptions.allow_concurrent_callbacks默认情况下为 false。因此您需要定义自己的SubscribeOptions。以下是您需要订阅并允许并发回调调用的代码：

ros::SubscribeOptions ops;
ops.template init<std_msgs::String>("chatter", 1000, chatterCallback);
ops.transport_hints = ros::TransportHints();
ops.allow_concurrent_callbacks = true;
ros::Subscriber sub = nh.subscribe(ops);