组件（Component）

执行器+组件–>实现进程内****节点之间的零拷贝通信
共享内存–>同一主机，不同进程的节点之间的零拷贝通信

目的：让多个节点在同一个进程中运行以提升效率
解释：没有 main() 函数的 ROS 节点
组件 (Component)：不是一个可执行文件，而是一个共享库（在 Linux 下是 .so 文件，Windows 下是 .dll 文件）。无法独立运行，需要被一个宿主进程加载才能工作。宿主进程就是组件容器（Component Container）

部署的灵活性：采用组件后，进程布局（Process Layout）的决定被推迟到了部署（Deploy-time）阶段

• 一个组件放在不同的容器甚至独立的进程中运行，以获得更好的故障隔离（一个节点崩溃不影响其他节点）和调试便利性
• 多个组件放入同一个容器进程，享受高效的进程内通信（Intra-Process Communication） 和更低的内存占用

如何实现高效的进程内通信？

1. 巧妙地绕过了ROS 2底层的DDS（数据分发服务）协议栈—拦截通信路径
   传统的独立进程节点通信：
   发送方：将数据序列化成二进制流 → 交给DDS中间件 → DDS通过网卡/共享内存发送。
   接收方：DDS接收数据 → 反序列化还原成对象 → 交给节点使用。
   这个过程涉及序列化（Serialize）和反序列化（Deserialize），即使是同一台机器上的两个进程，也需要走一遍DDS的完整协议栈，CPU开销很大。
   而当两个组件在同一个容器进程内时，ROS 2的中间层（rclcpp）会检测到它们的发布-订阅关系发生在同一进程。
   此时，数据流被截胡了：
   发送方：照常调用publish()。
   ROS 2中间件：检测到订阅者在同一进程，不再调用DDS的write()函数，而是直接将包含数据的智能指针（unique_ptr）传给订阅者。
   接收方：订阅者的回调函数直接拿到这个智能指针，无需任何序列化和反序列化操作。
2. 零拷贝（Zero-Copy）传输—共享内存
   传递的是指向这块内存的指针（引用），而不是数据本身。
   传统方式：数据被复制多次（应用内存 → DDS发送缓冲区 → DDS接收缓冲区 → 应用内存）。
   进程内通信：数据从发送端“移交”（move）给接收端。发布者发布后，它不再拥有这块数据的所有权，所有权直接转移给订阅者。整个过程没有一次数据内存的拷贝，延迟可以降到微秒级。
3. 共享的执行器（Executor）与线程模型
   当你把组件加载进容器时，它们共享容器的执行器（如SingleThreadedExecutor或MultiThreadedExecutor）。
   单线程（component_container）：所有组件的回调（定时器、订阅回调）都在同一个线程中按顺序执行。这避免了多线程环境下的互斥锁（Mutex）竞争和上下文切换（Context Switch）开销。因为数据是顺序处理的，不需要加锁保护共享资源，进一步减少了CPU时间浪费。
   多线程（component_container_mt）：即使使用多线程，由于数据是通过指针直接传递，也避免了线程间拷贝大块内存的开销。
4. 内存占用低
   代码段共享：容器进程加载了libtalker_component.so和liblistener_component.so。这些共享库的代码段（.text）在物理内存中只有一份。如果你用10个独立进程，每个进程都要把相同的ROS 2核心库和组件代码加载一遍；
   而用容器，所有组件共用容器的进程空间，避免了重复加载基础库。
   堆内存共享：如果是在不同进程，发布者存一份，DDS存一份，订阅者又存一份，内存占用会成倍增长。由于数据是零拷贝的，一份图像数据（比如1MB）只占用一次内存。

使用组件和容器

加载到一个特殊的宿主进程——组件容器（Component Container）里运行

示例组件

ros2 component types
--------------------
composition
  composition::Talker
  composition::Listener
  composition::NodeLikeListener
  composition::Server
  composition::Client

创建容器

ros2 run rclcpp_components component_container

查看容器

######################
ros2 component list 
-------------------
/ComponentManager
/listener
/talker
#######################
ros2 node list
-------------------
/ComponentManager
  1  /talker
  2  /listener

添加组件—使用组件管理器

ros2 component load /ComponentManager composition composition::Talker -e use_intra_process:=true
ros2 component load /ComponentManager composition composition::Listener -e use_intra_process:=true

ros2 component load /ComponentManager composition composition::Server 
ros2 component load /ComponentManager composition composition::Client 

此时使用的是进程内通信

自定义组件

1.创建节点类—和平常一模一样

公开继承自 rclcpp::Node

// my_component.hpp
#include "rclcpp/rclcpp.hpp"

namespace my_package {
class MyComponent : public rclcpp::Node {
public:
    // 构造函数必须接收 rclcpp::NodeOptions 参数
    explicit MyComponent(const rclcpp::NodeOptions & options);
private:

};
}

2.注册组件

在对应的 .cpp 文件中，使用 rclcpp_components 包提供的宏将你的类注册为组件

#include "my_component.hpp"
#include "rclcpp_components/register_node_macro.hpp"

RCLCPP_COMPONENTS_REGISTER_NODE(my_package::MyComponent)

3.构建共享库

组件不是可执行文件，是动态链接库
在CMakeLists.txt 中，将源文件编译为共享库，并使用专门的宏将其注册到 ROS 2 的组件索引中，以便 ros2 component 命令行工具能够发现它

find_package(rclcpp_components REQUIRED)
add_library(my_component SHARED
  src/my_component.cpp
)
rclcpp_components_register_nodes(my_component "my_package::MyComponent")    # 将 my_package::MyComponent注册为可发现的组件
ament_target_dependencies(my_component rclcpp rclcpp_components)

使用组件

launch文件启动

def generate_launch_description():
    # 1. 定义第一个组件
    cam2image_node = ComposableNode(
        package='image_tools',
        plugin='image_tools::Cam2Image',
        name='cam2image',
        # 可以配置参数、重映射等
        parameters=[{'width': 320, 'height': 240}],
        extra_arguments=[{'use_intra_process_comms': True}],
    )
    
    # 2. 定义第二个组件
    showimage_node = ComposableNode(
        package='image_tools',
        plugin='image_tools::ShowImage',
        name='showimage',
    )

    # 3. 创建容器，并将两个组件放入其中
    container = ComposableNodeContainer(
        name='my_image_container',
        package='rclcpp_components',
        executable='component_container_mt',  # 使用多线程容器
        composable_node_descriptions=[cam2image_node, showimage_node],
        output='screen',
    )

    return LaunchDescription([container])

编程集成—多线程执行器

在代码里直接实例化多个组件类，并在一个进程中运行
将多个节点放入多线程执行器当中

组件管理器（ComponentManager）

rclcpp_components::ComponentManager

自定义的组件管理器（IsolatedComponentManager）

/**
 * @brief 自定义 ComponentManager —— 双执行器分流模式
 *
 * 通过 LoadNode 服务动态加载的组件, 容器只能拿到基类接口, 无法调用组件的
 * 自定义方法 (如 get_pub_group())。因此按【回调组类型】进行执行器分流:
 *
 * 组件端建组约定:
 * - 话题/发布类回调 -> MutuallyExclusive 组
 * - 服务/动作类回调 -> Reentrant 组
 *
 * 容器端分流:
 * - MutuallyExclusive 组 -> 话题执行器 (StaticSingleThreadedExecutor, 稳态不重建
 *   wait_set, 适合高频 timer 发布, 固定单线程)
 * - Reentrant 组         -> 服务/动作执行器 (线程数可配: <=1 用 SingleThreaded,
 *   >=2 用 MultiThreaded)
 */
class IsolatedComponentManager : public rclcpp_components::ComponentManager
{
public:
    using rclcpp_components::ComponentManager::ComponentManager;

    ~IsolatedComponentManager() override
    {
        std::vector<std::shared_ptr<ExecutorWrapper>> wrappers;
        {
            std::lock_guard<std::mutex> lk(wrappers_mtx_);
            for (auto& kv : executor_wrappers_)
            {
                wrappers.push_back(kv.second);
            }
        }

        for (auto& w : wrappers)
        {
            if (w->topic_executor)
            {
                w->topic_executor->cancel();
            }
            if (w->service_action_executor)
            {
                w->service_action_executor->cancel();
            }
        }

        for (auto& w : wrappers)
        {
            if (w->topic_spin_thread.joinable())
            {
                w->topic_spin_thread.join();
            }
            if (w->service_action_spin_thread.joinable())
            {
                w->service_action_spin_thread.join();
            }
        }

        node_wrappers_.clear();
    }

    void set_service_action_thread_num_for_next_load(uint32_t n)
    {
        pending_service_action_thread_num_ = (n == 0U ? 1U : n);
    }

protected:
    void add_node_to_executor(uint64_t node_id) override
    {
        const uint32_t sa_threads = pending_service_action_thread_num_;
        pending_service_action_thread_num_ = 1U;

        auto wrapper = std::make_shared<ExecutorWrapper>();
        wrapper->topic_executor = std::make_shared<rclcpp::executors::StaticSingleThreadedExecutor>();

        if (sa_threads <= 1U)
        {
            wrapper->service_action_executor = std::make_shared<rclcpp::executors::SingleThreadedExecutor>();
        }
        else
        {
            wrapper->service_action_executor = std::make_shared<rclcpp::executors::MultiThreadedExecutor>(
                rclcpp::ExecutorOptions(), sa_threads);
        }

        auto node_base = node_wrappers_[node_id].get_node_base_interface();

        // ------------------------------------------------------------------
        // 关键: 若加载的组件是 LifecycleNode, 其 pub/sub/timer/service 等实体
        // 通常在 on_activate() 才创建。若在未激活时就把回调组交给执行器 spin,
        // StaticSingleThreadedExecutor 会缓存一个尚未包含真实实体(或半初始化
        // QoS event handler)的 wait_set, 导致运行期访问失效实体而 SIGSEGV。
        //
        // 因此在分流回调组、启动 spin 之前, 主动把组件驱动到 active 状态,
        // 确保所有实体与回调组关系创建完成, 再交给执行器。
        //
        // 注意: get_node_instance() 返回 shared_ptr<void>(类型擦除指针),
        // 不能直接 dynamic_pointer_cast(void 无 RTTI)。本容器约定加载的组件
        // 均为 LifecycleNode, 因此先用 node_base 判定后再 static_pointer_cast。
        // ------------------------------------------------------------------
        auto lifecycle_node = try_cast_lifecycle(node_id);
        if (lifecycle_node)
        {
            const auto & state = lifecycle_node->get_current_state();
            if (state.id() == lifecycle_msgs::msg::State::PRIMARY_STATE_UNCONFIGURED)
            {
                const auto & cfg_state = lifecycle_node->configure();
                if (cfg_state.id() != lifecycle_msgs::msg::State::PRIMARY_STATE_INACTIVE)
                {
                    LOGE("Component node_id=%lu configure failed, current state: %s",
                         node_id, cfg_state.label().c_str());
                    return;
                }
            }
            if (lifecycle_node->get_current_state().id() ==
                lifecycle_msgs::msg::State::PRIMARY_STATE_INACTIVE)
            {
                const auto & act_state = lifecycle_node->activate();
                if (act_state.id() != lifecycle_msgs::msg::State::PRIMARY_STATE_ACTIVE)
                {
                    LOGE("Component node_id=%lu activate failed, current state: %s",
                         node_id, act_state.label().c_str());
                    return;
                }
            }
            LOGI("Component node_id=%lu driven to active state, start dispatching callback groups", node_id);
        }

        // 按回调组类型分流:
        // - Reentrant 组         -> 服务/动作执行器
        // - MutuallyExclusive 组 -> 话题执行器 (节点的 default 组也是此类型,
        //   因此未显式建组的节点会全部归到话题执行器, 保守单线程)
        node_base->for_each_callback_group(
            [&wrapper, &node_base](rclcpp::CallbackGroup::SharedPtr group)
            {
                if (!group)
                {
                    return;
                }

                if (group->type() == rclcpp::CallbackGroupType::Reentrant)
                {
                    wrapper->service_action_executor->add_callback_group(group, node_base);
                }
                else
                {
                    wrapper->topic_executor->add_callback_group(group, node_base);
                }
            });

        auto topic_exec = wrapper->topic_executor;
        auto sa_exec = wrapper->service_action_executor;

        wrapper->topic_spin_thread = std::thread([topic_exec]() {
            topic_exec->spin();
        });

        wrapper->service_action_spin_thread = std::thread([sa_exec]() {
            sa_exec->spin();
        });

        {
            std::lock_guard<std::mutex> lk(wrappers_mtx_);
            executor_wrappers_[node_id] = wrapper;
        }
    }

    void remove_node_from_executor(uint64_t node_id) override
    {
        std::shared_ptr<ExecutorWrapper> wrapper;
        {
            std::lock_guard<std::mutex> lk(wrappers_mtx_);
            auto it = executor_wrappers_.find(node_id);
            if (it == executor_wrappers_.end())
            {
                return;
            }
            wrapper = it->second;
            executor_wrappers_.erase(it);
        }

        if (wrapper->topic_executor)
        {
            wrapper->topic_executor->cancel();
        }
        if (wrapper->service_action_executor)
        {
            wrapper->service_action_executor->cancel();
        }

        if (wrapper->topic_spin_thread.joinable())
        {
            wrapper->topic_spin_thread.join();
        }
        if (wrapper->service_action_spin_thread.joinable())
        {
            wrapper->service_action_spin_thread.join();
        }

        // spin 已停止, 若组件是 LifecycleNode 则驱动其 deactivate, 销毁实体,
        // 与 add_node_to_executor 里的 activate 对称, 避免实体悬挂。
        auto lifecycle_node = try_cast_lifecycle(node_id);
        if (lifecycle_node &&
            lifecycle_node->get_current_state().id() ==
               lifecycle_msgs::msg::State::PRIMARY_STATE_ACTIVE)
        {
            lifecycle_node->deactivate();
            LOGI("Component node_id=%lu deactivated", node_id);
        }
    }

private:
    // 把 node_wrappers_ 里的类型擦除指针还原为 LifecycleNode。
    // get_node_instance() 返回 shared_ptr<void>, void 无 RTTI 不能 dynamic_cast,
    // 本容器约定加载组件均继承 rclcpp_lifecycle::LifecycleNode, 故用 static_pointer_cast。
    // 若 node_id 不存在或实例为空则返回 nullptr。
    std::shared_ptr<rclcpp_lifecycle::LifecycleNode> try_cast_lifecycle(uint64_t node_id)
    {
        auto it = node_wrappers_.find(node_id);
        if (it == node_wrappers_.end())
        {
            return nullptr;
        }
        auto instance = it->second.get_node_instance();
        if (!instance)
        {
            return nullptr;
        }
        return std::static_pointer_cast<rclcpp_lifecycle::LifecycleNode>(instance);
    }

    struct ExecutorWrapper
    {
        rclcpp::Executor::SharedPtr topic_executor;
        rclcpp::Executor::SharedPtr service_action_executor;
        std::thread topic_spin_thread;
        std::thread service_action_spin_thread;
    };

    uint32_t pending_service_action_thread_num_{1U};
    std::mutex wrappers_mtx_;
    std::unordered_map<uint64_t, std::shared_ptr<ExecutorWrapper>> executor_wrappers_;
};

组件容器（Component Container）

是可执行程序（进程），宿主进程，提供执行器 Executor，运行多个组件；容器内部实例化一个ComponentManager节点提供服务接口
容器 = 进程 + 执行器 + 组件管理器