📁 项目结构

文本

编辑

/home/user/ros_docker_demo/
├── Dockerfile                  # 容器镜像
├── publisher_node.py           # 容器内发布节点（发布节点名+时间）
├── start_containers.sh         # 一键启动脚本
├── host_subscriber.py          # 宿主机订阅验证
└── stop_containers.sh          # 停止清理脚本

1️⃣ 容器发布节点（publisher_node.py）

python

编辑

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
容器内ROS发布节点
发布内容：节点名称 + 当前时间
"""

import rospy
import time
from std_msgs.msg import String
import os
from datetime import datetime

class PublisherNode:
    def __init__(self):
        # 从环境变量获取节点配置
        self.node_name = os.environ.get('NODE_NAME', 'default_node')
        self.container_id = os.environ.get('CONTAINER_ID', '0')
        
        # 初始化ROS节点
        rospy.init_node(self.node_name, anonymous=False)
        
        # 创建发布者（话题名包含容器ID，避免冲突）
        topic_name = f'/container_{self.container_id}/data'
        self.pub = rospy.Publisher(topic_name, String, queue_size=10)
        
        rospy.loginfo(f"✅ 节点启动: {self.node_name}")
        rospy.loginfo(f"📤 发布话题: {topic_name}")
        
    def run(self):
        rate = rospy.Rate(1)  # 1Hz，每秒发布一次
        
        while not rospy.is_shutdown():
            # 获取当前时间
            current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            
            # 构造消息内容
            message = f"节点: {self.node_name} | 时间: {current_time}"
            
            # 发布消息
            self.pub.publish(message)
            
            # 打印日志
            rospy.loginfo(f"📤 已发布: {message}")
            
            rate.sleep()

if __name__ == '__main__':
    try:
        node = PublisherNode()
        node.run()
    except rospy.ROSInterruptException:
        pass

2️⃣ Dockerfile

dockerfile

编辑

FROM ros:noetic

# 设置环境变量
ENV DEBIAN_FRONTEND=noninteractive

# 安装依赖
RUN apt-get update && apt-get install -y \
    python3-pip \
    && rm -rf /var/lib/apt/lists/*

# 复制发布节点代码
COPY publisher_node.py /opt/ros_publisher/

# 设置工作目录
WORKDIR /opt/ros_publisher

# 默认命令
CMD ["bash", "-c", "source /opt/ros/noetic/setup.bash && python3 /opt/ros_publisher/publisher_node.py"]

3️⃣ 一键启动脚本（start_containers.sh）

bash

编辑

#!/bin/bash
# ============================================
# 双容器节点启动脚本（无相机版本）
# ============================================

set -e

# 配置
IMAGE_NAME="ros_publisher:v1"
CURRENT_TIME=$(date +%Y%m%d_%H%M%S)

echo "=========================================="
echo "  双容器节点启动脚本（无相机版）"
echo "  当前时间: $CURRENT_TIME"
echo "=========================================="

# 1. 检查 Docker 镜像是否存在
if ! docker images | grep -q "$IMAGE_NAME"; then
    echo "❌ 镜像不存在，正在构建..."
    docker build -t $IMAGE_NAME .
fi

# 2. 清理旧容器
echo "🧹 清理旧容器..."
docker rm -f publisher_container_0 2>/dev/null || true
docker rm -f publisher_container_1 2>/dev/null || true

# 3. 启动容器1
echo "🚀 启动容器1..."
docker run -d \
  --name publisher_container_0 \
  --network host \
  -e NODE_NAME="container_0_node_${CURRENT_TIME}" \
  -e CONTAINER_ID=0 \
  $IMAGE_NAME

# 4. 启动容器2
echo "🚀 启动容器2..."
docker run -d \
  --name publisher_container_1 \
  --network host \
  -e NODE_NAME="container_1_node_${CURRENT_TIME}" \
  -e CONTAINER_ID=1 \
  $IMAGE_NAME

# 5. 等待容器启动
sleep 3

# 6. 验证
echo ""
echo "=========================================="
echo "  验证步骤"
echo "=========================================="
echo "1. 确保宿主机已运行 roscore"
echo "   命令: roscore"
echo ""
echo "2. 查看ROS节点列表"
echo "   命令: rosnode list"
echo ""
echo "3. 查看话题列表"
echo "   命令: rostopic list"
echo ""
echo "4. 订阅容器0数据"
echo "   命令: rostopic echo /container_0/data"
echo ""
echo "5. 订阅容器1数据"
echo "   命令: rostopic echo /container_1/data"
echo ""
echo "6. 查看容器日志"
echo "   命令: docker logs -f publisher_container_0"
echo "   命令: docker logs -f publisher_container_1"
echo ""
echo "7. 运行宿主机订阅程序"
echo "   命令: python3 host_subscriber.py"
echo ""
echo "=========================================="
echo "  ✅ 启动完成！"
echo "=========================================="

4️⃣ 宿主机订阅验证脚本（host_subscriber.py）

python

编辑

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
宿主机订阅节点 - 验证容器数据回传
"""

import rospy
from std_msgs.msg import String
from datetime import datetime

class HostSubscriber:
    def __init__(self):
        rospy.init_node('host_subscriber', anonymous=True)
        
        self.data_count_0 = 0
        self.data_count_1 = 0
        self.last_data_0 = ""
        self.last_data_1 = ""
        
    def callback_0(self, msg):
        self.data_count_0 += 1
        self.last_data_0 = msg.data
        rospy.loginfo(f"✅ [容器0] 收到第 {self.data_count_0} 条数据: {msg.data}")
            
    def callback_1(self, msg):
        self.data_count_1 += 1
        self.last_data_1 = msg.data
        rospy.loginfo(f"✅ [容器1] 收到第 {self.data_count_1} 条数据: {msg.data}")
    
    def run(self):
        # 订阅两个容器的话题
        rospy.Subscriber('/container_0/data', String, self.callback_0)
        rospy.Subscriber('/container_1/data', String, self.callback_1)
        
        rospy.loginfo("🎯 宿主机订阅节点已启动，等待容器数据...")
        rospy.loginfo("📋 订阅话题: /container_0/data, /container_1/data")
        
        rate = rospy.Rate(5)  # 每5秒打印一次统计
        while not rospy.is_shutdown():
            rospy.loginfo("========================================")
            rospy.loginfo(f"📊 数据统计:")
            rospy.loginfo(f"   容器0: {self.data_count_0} 条 | 最新: {self.last_data_0}")
            rospy.loginfo(f"   容器1: {self.data_count_1} 条 | 最新: {self.last_data_1}")
            rospy.loginfo("========================================")
            rate.sleep()
        
        rospy.loginfo("👋 订阅节点退出")

if __name__ == '__main__':
    try:
        node = HostSubscriber()
        node.run()
    except rospy.ROSInterruptException:
        pass

5️⃣ 停止清理脚本（stop_containers.sh）

bash

编辑

#!/bin/bash
# ============================================
# 停止并清理容器
# ============================================

echo "🛑 停止容器..."
docker stop publisher_container_0 2>/dev/null || true
docker stop publisher_container_1 2>/dev/null || true

echo "🧹 删除容器..."
docker rm publisher_container_0 2>/dev/null || true
docker rm publisher_container_1 2>/dev/null || true

echo "✅ 清理完成！"

6️⃣ 完整执行流程

步骤1：创建项目目录并保存文件

bash

编辑

# 创建目录
mkdir -p /home/user/ros_docker_demo
cd /home/user/ros_docker_demo

# 将上述代码分别保存为对应文件
# publisher_node.py
# Dockerfile
# start_containers.sh
# host_subscriber.py
# stop_containers.sh

# 给脚本添加执行权限
chmod +x start_containers.sh stop_containers.sh

步骤2：构建 Docker 镜像

bash

编辑

cd /home/user/ros_docker_demo
docker build -t ros_publisher:v1 .

步骤3：在宿主机启动 ROS Master

bash

编辑

# 打开终端1
roscore

步骤4：启动两个容器

bash

编辑

# 打开终端2
cd /home/user/ros_docker_demo
./start_containers.sh

步骤5：验证数据回传

bash

编辑

# 方法A：查看节点列表
rosnode list
# 应该看到类似：
# /container_0_node_20260311_093600
# /container_1_node_20260311_093600
# /host_subscriber

# 方法B：查看话题列表
rostopic list
# 应该看到：
# /container_0/data
# /container_1/data

# 方法C：订阅数据验证
rostopic echo /container_0/data
rostopic echo /container_1/data

# 方法D：运行宿主机订阅程序
python3 host_subscriber.py

步骤6：查看容器日志

bash

编辑

docker logs -f publisher_container_0
docker logs -f publisher_container_1

步骤7：停止清理

bash

编辑

./stop_containers.sh

7️⃣ 预期输出示例

容器日志输出

文本

编辑

[INFO] 节点启动: container_0_node_20260311_093600
[INFO] 发布话题: /container_0/data
[INFO] 已发布: 节点: container_0_node_20260311_093600 | 时间: 2026-03-11 09:36:01
[INFO] 已发布: 节点: container_0_node_20260311_093600 | 时间: 2026-03-11 09:36:02
...

宿主机订阅输出

文本

编辑

[INFO] 宿主机订阅节点已启动，等待容器数据...
[INFO] ✅ [容器0] 收到第 1 条数据: 节点: container_0_node_20260311_093600 | 时间: 2026-03-11 09:36:01
[INFO] ✅ [容器1] 收到第 1 条数据: 节点: container_1_node_20260311_093600 | 时间: 2026-03-11 09:36:01
...

8️⃣ 极简快速测试版（无需构建镜像）

如果只想快速测试，可以直接使用官方 ROS 镜像 + 内联命令：

bash

编辑

# 1. 宿主机启动 roscore
roscore

# 2. 获取当前时间戳
TIME_STAMP=$(date +%Y%m%d_%H%M%S)

# 3. 启动容器1
docker run -d \
  --name test_container_0 \
  --network host \
  ros:noetic \
  bash -c "source /opt/ros/noetic/setup.bash && \
    rosrun rospy_tutorials talker __name:=node_${TIME_STAMP}_0"

# 4. 启动容器2
docker run -d \
  --name test_container_1 \
  --network host \
  ros:noetic \
  bash -c "source /opt/ros/noetic/setup.bash && \
    rosrun rospy_tutorials talker __name:=node_${TIME_STAMP}_1"

# 5. 宿主机验证
rosnode list
rostopic echo /chatter

9️⃣ 常见问题排查

表格

问题检查命令解决方案
容器启动后立即退出docker logs publisher_container_0检查 Python 代码错误
rosnode list 看不到节点docker logs publisher_container_0确认 roscore 在运行
收不到数据rostopic list确认话题名正确
节点名重复-每次启动使用不同时间戳

✅ 验证成功的标志

bash

编辑

# 1. 节点列表显示两个容器节点（带时间戳）
$ rosnode list
/container_0_node_20260311_093600
/container_1_node_20260311_093600

# 2. 话题列表显示两个数据话题
$ rostopic list
/container_0/data
/container_1/data

# 3. 宿主机能收到数据
$ rostopic echo /container_0/data
data: "节点: container_0_node_20260311_093600 | 时间: 2026-03-11 09:36:01"
---
data: "节点: container_0_node_20260311_093600 | 时间: 2026-03-11 09:36:02"

📋 核心配置总结

表格

配置项值说明
网络模式--network host容器与宿主机共享网络
节点命名container_X_node_时间戳每次启动唯一
话题命名/container_X/data区分不同容器
发布频率1Hz每秒一条数据
数据内容节点名 + 当前时间验证数据来源

按照以上步骤操作，无需相机即可验证双容器 + 数据回传宿主机！ 🎉

# docker
Logo
亚马逊云科技技术品牌专区

更多推荐

open-code-review Codex 插件教程：对话式调用评审技能指南

本文是一份面向新手的 **open-code-review Codex 插件教程**，教你在 Codex 中通过对话式方式调用 open-code-review 的 AI 代码评审技能。open-code-review 是阿里巴巴开源的 AI 代码评审工具，采用"确定性流水线 + LLM Agent"混合架构，内置多语言评审规则（NPE、线程安全、XSS、SQL 注入等），可在 Codex 里用一

avatar 亚马逊云科技技术品牌专区

OBS Studio 完整指南：免费开源录屏与直播软件从零上手

想做一场像模像样的直播，或者把屏幕内容录成教程视频，但一想到商用软件的订阅费就有些犹豫？其实你不需要花一分钱。OBS Studio 是一款免费开源的直播与录屏软件，支持 Windows、macOS 和 Linux，采集、混音、编码、推流、录制，一条流水线全包了。## 先搞清楚 OBS Studio 是什么OBS Studio 是一个"场景化"的采集合成工具：你把摄像头、窗口、浏览器、图片、

avatar 亚马逊云科技技术品牌专区

如何用MediaCrawler爬取小红书抖音数据？7大平台爬虫从安装到实战完整指南

MediaCrawler 是一个支持小红书、抖音、快手、B站、微博、百度贴吧、知乎的开源自媒体数据采集工具。它能按关键词、按帖子、按创作者主页三种方式抓取笔记/视频内容、评论甚至二级评论，并把结果存成 JSON、CSV、Excel 或数据库。适合做市场调研、竞品监控、内容选题的运营人员，以及想要现成爬虫框架而不想从零写 JS 逆向的开发者。## 从一次"手动抄数据"说起想象这个场景：你负责

avatar 亚马逊云科技技术品牌专区