Topic 실습 — Publisher와 Subscriber부터 실제 로봇까지

학습 목표

Topic이 비동기 Publish·Subscribe 통신이며 응답이나 처리 완료를 보장하지 않는 이유를 설명할 수 있다.
Topic 이름, Message Type과 QoS라는 세 가지 연결 조건을 점검할 수 있다.
ros2 topic과 관련 CLI 명령으로 값, 연결 수, 주기, 대역폭과 QoS를 조사할 수 있다.
Python Package를 만들고 Publisher와 Subscriber를 작성·Build·실행할 수 있다.
Queue Depth, Callback 처리 시간과 Best Effort 유실의 차이를 진단할 수 있다.
Remapping과 Namespace로 같은 Code를 여러 Robot에 재사용할 수 있다.
/cmd_vel, /scan, /odom이 실제 이동 Robot에서 어떻게 연결되는지 설명할 수 있다.
Deadman Timeout, 속도 제한과 장애물 정지를 포함한 안전 Filter를 구현할 수 있다.

1. Topic이 ROS에서 가장 중요한 이유

Robot은 끊임없이 상태를 측정하고 명령을 갱신합니다. Camera Image, LiDAR Scan, Wheel Odometry, Battery State, Joint State와 속도 명령은 한 번 요청하고 끝나는 Data가 아니라 시간에 따라 계속 흐르는 Stream입니다.

ROS Topic은 이 Stream을 전달하는 기본 통신 방식입니다.

Sensor Driver ── Message Stream ──> Topic ──> Perception
                                        ├──> Logger
                                        └──> Visualization

Publisher는 정해진 Topic에 Message를 발행하고 Subscriber는 필요한 Topic을 구독합니다. 서로의 Class, Process ID, IP 주소나 Programming Language를 직접 알 필요가 없습니다.

이 느슨한 결합은 다음을 가능하게 합니다.

실제 LiDAR 대신 Bag Replay를 연결
C++ Driver와 Python Algorithm을 연결
기존 Code 수정 없이 Logger나 RViz를 추가
같은 Algorithm을 Simulation과 실제 Robot에서 재사용
한 Publisher Data를 여러 Subscriber가 동시에 사용

그러나 Topic은 요청·응답이 아닙니다. Publisher는 누가 받았는지, 처리에 성공했는지 알지 못합니다. “Map을 저장하고 완료 여부를 알려 달라”면 Service, “목표까지 이동하며 Feedback과 Cancel이 필요하다”면 Action이 더 적합합니다.

2. Topic 연결의 세 가지 조건

Publisher와 Subscriber는 다음 세 조건이 맞아야 Data를 교환합니다.

Topic 이름 + Message Type + 호환되는 QoS

Topic 이름

/scan과 /robot_01/scan은 다른 Topic입니다. 철자, 대소문자와 Namespace가 정확해야 합니다.

Message Type

geometry_msgs/msg/Twist와 geometry_msgs/msg/TwistStamped는 Field가 비슷해도 다른 Type입니다. 이름이 같아도 연결되지 않습니다.

QoS

Subscriber가 Reliable을 요구하지만 Publisher가 Best Effort만 제공하면 연결되지 않을 수 있습니다. 이름과 Type이 같아도 Callback이 호출되지 않는 대표적인 원인입니다.

Publisher는 Subscriber가 0명이어도 오류 없이 계속 발행하고 Subscriber는 Publisher가 없어도 계속 기다립니다. “상대가 없다”는 상황이 예외가 아니라 정상 상태이므로 CLI로 연결을 확인해야 합니다.

3. 실습 환경 준비

모든 Terminal에서 설치한 ROS 2 환경을 불러옵니다.

source /opt/ros/<distro>/setup.bash
printenv ROS_DISTRO

<distro>는 설치된 배포판 이름으로 바꿉니다. 사용자 Workspace를 만들었다면 그 뒤에 Overlay를 Source합니다.

source /opt/ros/<distro>/setup.bash
source ~/topic_ws/install/setup.bash

환경이 정상인지 확인합니다.

ros2 --help
ros2 doctor --report
ros2 pkg executables demo_nodes_py

여러 Computer에서 실습한다면 같은 ROS_DOMAIN_ID를 사용하고 Discovery가 가능한 Network인지 확인합니다.

echo "$ROS_DOMAIN_ID"
export ROS_DOMAIN_ID=21

Domain ID는 보안 경계가 아닙니다. 다른 Domain의 자동 발견을 분리하는 논리적 구분 값입니다.

4. 첫 실습 — Code 없이 문자열 발행·구독

Terminal 1: Subscriber

ros2 topic echo /practice/chatter std_msgs/msg/String

아직 Publisher가 없으므로 아무것도 출력되지 않는 것이 정상입니다.

Terminal 2: 한 번 발행

ros2 topic pub --once \
  /practice/chatter std_msgs/msg/String \
  "{data: '안녕하세요 ROS 2 Topic'}"

2 Hz로 계속 발행

ros2 topic pub --rate 2 \
  /practice/chatter std_msgs/msg/String \
  "{data: '2 Hz message'}"

지정 횟수만 발행

ros2 topic pub --times 5 --rate 2 \
  /practice/chatter std_msgs/msg/String \
  "{data: 'five messages'}"

ros2 topic pub의 Message 값은 YAML 문법입니다. Shell Quote와 YAML Brace가 함께 쓰이므로 전체 Data를 큰따옴표로 묶는 습관이 안전합니다.

5. Topic 조사 명령 전체 흐름

Topic 이름과 Type 찾기

ros2 topic list
ros2 topic list -t
ros2 topic type /practice/chatter
ros2 topic find std_msgs/msg/String

Message 구조 확인

ros2 interface show std_msgs/msg/String
ros2 interface show geometry_msgs/msg/Twist
ros2 interface show sensor_msgs/msg/LaserScan
ros2 interface show nav_msgs/msg/Odometry

연결 수와 Endpoint 확인

ros2 topic info /practice/chatter
ros2 topic info /practice/chatter --verbose

--verbose 결과에서 다음을 확인합니다.

Publisher와 Subscription 수
Node 이름과 Namespace
Topic Type과 Type Hash
Endpoint GID
Reliability, Durability, History와 Depth

Message 값 확인

ros2 topic echo /practice/chatter
ros2 topic echo /practice/chatter --once
ros2 topic echo /cmd_vel --field linear.x

주기와 대역폭 측정

ros2 topic hz /scan
ros2 topic bw /camera/image_raw

이 값은 CLI 명령이 만든 Subscriber에서 실제로 수신한 값입니다. Publisher 내부의 발행 횟수와 정확히 같다고 보장되지 않으며 CPU 부하, Network와 QoS 영향을 받습니다.

CLI Graph Cache가 의심될 때

ros2 daemon status
ros2 daemon stop
ros2 daemon start

CLI Daemon은 조회 속도를 높이는 Cache이며 ROS 2 통신의 중앙 Master가 아닙니다.

6. Twist 명령을 안전하게 읽고 발행하기

geometry_msgs/msg/Twist는 선속도 Vector와 각속도 Vector를 가집니다.

ros2 interface show geometry_msgs/msg/Twist

일반적인 평면 차동 구동 Robot은 다음 두 Field를 사용합니다.

linear.x: 전진 속도, 보통 m/s
angular.z: Z축 주위 Yaw Rate, 보통 rad/s

정지 Message 한 번 발행

ros2 topic pub --once /cmd_vel geometry_msgs/msg/Twist \
  "{linear: {x: 0.0}, angular: {z: 0.0}}"

Simulation에서 저속 전진

ros2 topic pub --rate 10 /cmd_vel geometry_msgs/msg/Twist \
  "{linear: {x: 0.10}, angular: {z: 0.0}}"

제자리 좌회전

ros2 topic pub --rate 10 /cmd_vel geometry_msgs/msg/Twist \
  "{linear: {x: 0.0}, angular: {z: 0.25}}"

실제 Robot에 /cmd_vel을 직접 발행하기 전에 Wheel을 바닥에서 띄우거나 충분한 안전 공간을 확보하고 Emergency Stop을 준비해야 합니다. Robot마다 Topic 이름, 단위, Watchdog와 허용 속도가 다르므로 Driver 문서를 먼저 확인하십시오. 위 값도 모든 Robot에 안전하다고 보장되지 않습니다.

명령을 중단할 때 Ctrl+C만 믿지 말고 별도 Terminal에서 정지 Message를 보내고 실제 정지를 확인합니다.

ros2 topic pub --once /cmd_vel geometry_msgs/msg/Twist "{}"

상용 Robot Driver는 Command가 끊기면 Hardware 또는 하위 Controller에서 Timeout으로 정지해야 합니다. 상위 CLI만으로 안전을 보장하면 안 됩니다.

7. Python Package 전체 만들기

Workspace와 Package 생성

mkdir -p ~/topic_ws/src
cd ~/topic_ws/src
ros2 pkg create topic_practice \
  --build-type ament_python \
  --license Apache-2.0 \
  --dependencies rclpy std_msgs geometry_msgs sensor_msgs nav_msgs

생성 구조는 다음과 같습니다.

topic_ws/src/topic_practice/
├── package.xml
├── resource/topic_practice
├── setup.cfg
├── setup.py
├── topic_practice/
│   └── __init__.py
└── test/

앞으로 작성할 File은 topic_practice/topic_practice/ 안에 둡니다.

`setup.py` 실행 Entry Point

from setuptools import find_packages, setup

package_name = 'topic_practice'

setup(
    name=package_name,
    version='0.1.0',
    packages=find_packages(exclude=['test']),
    data_files=[
        ('share/ament_index/resource_index/packages',
         ['resource/' + package_name]),
        ('share/' + package_name, ['package.xml']),
    ],
    install_requires=['setuptools'],
    zip_safe=True,
    maintainer='Robossam Student',
    maintainer_email='student@example.com',
    description='ROS 2 topic publisher and subscriber practice',
    license='Apache-2.0',
    entry_points={
        'console_scripts': [
            'talker = topic_practice.talker:main',
            'listener = topic_practice.listener:main',
            'cmd_vel_filter = topic_practice.cmd_vel_filter:main',
            'odom_monitor = topic_practice.odom_monitor:main',
        ],
    },
)

Dependency 설치와 Build

cd ~/topic_ws
rosdep install --from-paths src --ignore-src -r -y
colcon build --symlink-install --packages-select topic_practice
source install/setup.bash
ros2 pkg executables topic_practice

Python Source를 바꿀 때 --symlink-install이면 즉시 반영되는 경우가 많지만 setup.py, Entry Point나 Interface를 바꾸면 다시 Build해야 합니다.

8. Publisher 프로그램 — Talker

topic_practice/talker.py를 작성합니다.

import rclpy
from rclpy.node import Node
from std_msgs.msg import String


class Talker(Node):
    def __init__(self):
        super().__init__('my_talker')

        # Type, 상대 Topic 이름, Queue Depth
        self.publisher = self.create_publisher(String, 'chatter', 10)
        self.count = 0
        self.timer = self.create_timer(0.5, self.on_timer)  # 2 Hz

    def on_timer(self):
        message = String()
        message.data = f'hello #{self.count}'
        self.publisher.publish(message)
        self.get_logger().info(f'발행: {message.data}')
        self.count += 1


def main(args=None):
    rclpy.init(args=args)
    node = Talker()
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        pass
    finally:
        node.destroy_node()
        rclpy.shutdown()


if __name__ == '__main__':
    main()

chatter에 앞 Slash를 붙이지 않았으므로 상대 이름입니다. Namespace가 /robot_01이면 최종 Topic은 /robot_01/chatter가 됩니다.

cd ~/topic_ws
source install/setup.bash
ros2 run topic_practice talker

다른 Terminal에서 확인합니다.

ros2 node info /my_talker
ros2 topic echo /chatter
ros2 topic hz /chatter

9. Subscriber 프로그램 — Listener

topic_practice/listener.py를 작성합니다.

import rclpy
from rclpy.node import Node
from std_msgs.msg import String


class Listener(Node):
    def __init__(self):
        super().__init__('my_listener')
        self.received_count = 0

        # Type, Topic 이름, Callback, Queue Depth
        self.subscription = self.create_subscription(
            String,
            'chatter',
            self.on_message,
            10,
        )

    def on_message(self, message):
        self.received_count += 1
        self.get_logger().info(
            f'수신 #{self.received_count}: {message.data}')


def main(args=None):
    rclpy.init(args=args)
    node = Listener()
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        pass
    finally:
        node.destroy_node()
        rclpy.shutdown()


if __name__ == '__main__':
    main()

Subscription 객체를 self.subscription에 보관합니다. Callback은 Executor가 호출하므로 직접 호출하지 않습니다.

ros2 run topic_practice listener

Talker와 Listener를 실행한 상태에서 다음을 비교합니다.

ros2 topic info /chatter --verbose
rqt_graph

Listener를 두 개 실행하면 두 Process가 각각 Message 사본을 받습니다.

ros2 run topic_practice listener --ros-args -r __node:=listener_2

10. Remapping과 Namespace

Code에 Robot 이름을 고정하지 않고 실행할 때 바꿉니다.

Topic Remap

ros2 run topic_practice talker --ros-args \
  -r chatter:=status_text

ros2 run topic_practice listener --ros-args \
  -r chatter:=status_text

Namespace 적용

ros2 run topic_practice talker --ros-args \
  -r __ns:=/robot_01

ros2 run topic_practice listener --ros-args \
  -r __ns:=/robot_01

최종 Topic은 /robot_01/chatter가 됩니다.

두 Robot 분리

ros2 run topic_practice talker --ros-args \
  -r __node:=talker_01 -r __ns:=/robot_01

ros2 run topic_practice talker --ros-args \
  -r __node:=talker_02 -r __ns:=/robot_02

절대 Topic 이름 /chatter를 Code에 사용하면 Namespace가 앞에 붙지 않습니다. 재사용 가능한 Node는 보통 상대 이름을 사용하고 Launch에서 Namespace와 Remap을 정합니다.

11. QoS를 명시하는 Subscriber

LiDAR와 Camera Driver는 Sensor Data Profile을 사용하는 경우가 많습니다. 기본 Reliable Subscriber로는 Best Effort Publisher와 호환되지 않을 수 있습니다.

import rclpy
from rclpy.node import Node
from rclpy.qos import qos_profile_sensor_data
from sensor_msgs.msg import LaserScan


class ScanListener(Node):
    def __init__(self):
        super().__init__('scan_listener')
        self.subscription = self.create_subscription(
            LaserScan,
            'scan',
            self.on_scan,
            qos_profile_sensor_data,
        )

    def on_scan(self, message):
        valid = [distance for distance in message.ranges
                 if message.range_min <= distance <= message.range_max]
        if valid:
            self.get_logger().info(f'최단 거리: {min(valid):.3f} m')

CLI에서도 QoS를 맞춰 확인할 수 있습니다.

ros2 topic echo /scan sensor_msgs/msg/LaserScan \
  --qos-reliability best_effort \
  --qos-durability volatile

QoS 문제는 다음 순서로 봅니다.

ros2 topic type /scan
ros2 topic info /scan --verbose
ros2 topic echo /scan --qos-reliability best_effort

12. Queue Depth와 Message가 사라지는 세 경로

1. 연결 자체가 없음

Topic 이름 또는 Namespace 불일치
Message Type 불일치
QoS 불호환
다른 ROS Domain이나 Discovery 실패

2. Subscriber 처리 Queue가 밀림

20 Hz Message가 들어오는데 Callback이 한 번에 0.3초 걸리면 Callback 처리 속도는 약 3.3 Hz에 불과합니다. Keep Last Depth가 10이면 오래된 Message가 제거됩니다.

3. Best Effort Network 유실

재전송보다 최신 Data가 중요한 Sensor Stream은 Packet이 Network에서 유실될 수 있습니다.

Queue를 무조건 크게 만드는 것은 해결책이 아닙니다. 오래된 제어 명령이 쌓여 Robot이 과거를 뒤늦게 실행할 수 있습니다.

Data	일반적 우선순위	Queue 설계 방향
`/cmd_vel`	최신 명령과 낮은 지연	얕게, Driver Timeout 필수
`/scan`	최신 Scan	Sensor QoS, 얕게
Camera Image	최신 Frame, 대역폭	Best Effort와 얕은 Queue 검토
Map·고정 설정	늦게 참가해도 수신	Transient Local 검토
Event·감사 Log	누락 방지	Reliable과 별도 저장 전략

QoS와 Depth는 실제 주기, Message 크기, Network와 처리 시간을 측정해 결정합니다.

13. Callback에서 지켜야 할 원칙

기본 rclpy.spin(node)는 Single-threaded 실행입니다. Callback 하나가 오래 걸리면 다른 Subscription, Timer와 Service Callback도 기다립니다.

Callback 안에서 피해야 할 작업:

time.sleep()과 무기한 대기
큰 Image 추론이나 긴 Path Planning
느린 Disk·Network Blocking I/O
같은 Executor에서 처리되어야 할 Service를 동기 호출
Lock을 잡은 채 다른 Callback 결과 대기

권장 Pattern:

Callback에서 Message 검증과 최신 값 저장만 수행
Timer에서 일정한 출력 주기로 판단·발행
무거운 계산은 별도 Worker 또는 Node로 분리
필요할 때 Callback Group과 MultiThreadedExecutor를 설계
Callback 처리 시간과 Data Age를 측정

수신 주기가 정상이어도 처리 Data가 오래되었을 수 있습니다. Header.stamp가 있는 Message는 현재 ROS Time과 비교해 Age를 확인합니다.

14. 실제 Robot Program — 안전한 cmd_vel Filter

실제 Robot에서는 Navigation, Teleoperation과 Remote Control이 모두 속도 명령을 만들 수 있습니다. 이 명령을 Motor Driver에 바로 연결하면 다음 위험이 있습니다.

상위 Node나 Wi-Fi가 끊겨 마지막 속도로 계속 주행
허용 범위를 넘는 속도
LiDAR 앞에 장애물이 있는데 전진
여러 Publisher가 동시에 /cmd_vel에 발행해 명령이 충돌

아래 cmd_vel_filter.py는 /cmd_vel_raw와 /scan을 구독하고 20 Hz로 검증된 /cmd_vel을 발행합니다.

import math

import rclpy
from geometry_msgs.msg import Twist
from rclpy.node import Node
from rclpy.qos import qos_profile_sensor_data
from sensor_msgs.msg import LaserScan


class CmdVelFilter(Node):
    DEADMAN_SECONDS = 0.5
    MAX_LINEAR = 0.35
    MAX_ANGULAR = 1.0
    FRONT_STOP_DISTANCE = 0.40

    def __init__(self):
        super().__init__('cmd_vel_filter')

        self.command_subscription = self.create_subscription(
            Twist, 'cmd_vel_raw', self.on_command, 10)
        self.scan_subscription = self.create_subscription(
            LaserScan, 'scan', self.on_scan, qos_profile_sensor_data)
        self.safe_publisher = self.create_publisher(
            Twist, 'cmd_vel', 10)

        self.last_command = Twist()
        self.last_command_time = None
        self.front_distance = math.inf
        self.timer = self.create_timer(0.05, self.publish_safe_command)

    def on_command(self, message):
        # Callback은 최신 명령과 수신 시각만 저장하고 빨리 반환합니다.
        self.last_command = message
        self.last_command_time = self.get_clock().now()

    def on_scan(self, message):
        # 전방 ±20도 Sector만 사용합니다.
        half_angle = math.radians(20.0)
        valid = []
        for index, distance in enumerate(message.ranges):
            angle = message.angle_min + index * message.angle_increment
            if abs(angle) > half_angle:
                continue
            if message.range_min <= distance <= message.range_max:
                valid.append(distance)
        self.front_distance = min(valid) if valid else math.inf

    def publish_safe_command(self):
        output = Twist()  # 기본값 0 = 정지
        now = self.get_clock().now()

        if self.last_command_time is None:
            self.safe_publisher.publish(output)
            return

        age = (now - self.last_command_time).nanoseconds * 1e-9
        if age > self.DEADMAN_SECONDS:
            self.get_logger().warn(
                f'명령 Timeout {age:.2f}s: 정지',
                throttle_duration_sec=1.0)
            self.safe_publisher.publish(output)
            return

        linear = self.clamp(
            self.last_command.linear.x, self.MAX_LINEAR)
        angular = self.clamp(
            self.last_command.angular.z, self.MAX_ANGULAR)

        # 장애물이 가까우면 전진만 차단하고 뒤로 빠지거나 회전은 허용합니다.
        if self.front_distance <= self.FRONT_STOP_DISTANCE and linear > 0.0:
            linear = 0.0
            self.get_logger().warn(
                f'전방 장애물 {self.front_distance:.2f}m: 전진 차단',
                throttle_duration_sec=1.0)

        output.linear.x = linear
        output.angular.z = angular
        self.safe_publisher.publish(output)

    @staticmethod
    def clamp(value, limit):
        return max(-limit, min(limit, value))


def main(args=None):
    rclpy.init(args=args)
    node = CmdVelFilter()
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        pass
    finally:
        # 종료 직전 정지 Message를 발행하지만 Hardware Watchdog도 반드시 필요합니다.
        node.safe_publisher.publish(Twist())
        node.destroy_node()
        rclpy.shutdown()


if __name__ == '__main__':
    main()

Simulation에서 시험

ros2 run topic_practice cmd_vel_filter

# 다른 Terminal에서 Filter 입력으로 명령
ros2 topic pub --rate 10 /cmd_vel_raw geometry_msgs/msg/Twist \
  "{linear: {x: 0.15}, angular: {z: 0.0}}"

# 출력 확인
ros2 topic echo /cmd_vel

Publisher를 Ctrl+C로 종료하고 0.5초 뒤 /cmd_vel이 0으로 바뀌는지 확인합니다.

이 예제는 교육용 Application Filter입니다. LiDAR 고장, Process Crash나 Computer 전원 상실까지 막지 못합니다. 실제 Robot에는 Motor Controller의 Command Watchdog, Emergency Stop, Bumper와 위험 분석에 따른 독립 Safety Layer가 필요합니다.

15. 실제 Robot Program — Odometry 상태 감시

/odom은 Robot의 추정 위치와 속도를 담습니다. 다음 odom_monitor.py는 위치, Yaw, 속도와 Data Age를 출력합니다.

import math

import rclpy
from nav_msgs.msg import Odometry
from rclpy.node import Node


class OdomMonitor(Node):
    def __init__(self):
        super().__init__('odom_monitor')
        self.subscription = self.create_subscription(
            Odometry, 'odom', self.on_odom, 10)

    def on_odom(self, message):
        position = message.pose.pose.position
        orientation = message.pose.pose.orientation

        # Quaternion에서 Yaw 계산
        siny_cosp = 2.0 * (
            orientation.w * orientation.z +
            orientation.x * orientation.y)
        cosy_cosp = 1.0 - 2.0 * (
            orientation.y ** 2 + orientation.z ** 2)
        yaw = math.atan2(siny_cosp, cosy_cosp)

        stamp = rclpy.time.Time.from_msg(message.header.stamp)
        age = (self.get_clock().now() - stamp).nanoseconds * 1e-9
        linear = message.twist.twist.linear.x
        angular = message.twist.twist.angular.z

        self.get_logger().info(
            f'x={position.x:.2f}m y={position.y:.2f}m '
            f'yaw={yaw:.2f}rad v={linear:.2f}m/s '
            f'w={angular:.2f}rad/s age={age:.3f}s')


def main(args=None):
    rclpy.init(args=args)
    node = OdomMonitor()
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        pass
    finally:
        node.destroy_node()
        rclpy.shutdown()


if __name__ == '__main__':
    main()

실행과 비교 명령:

ros2 run topic_practice odom_monitor
ros2 topic hz /odom
ros2 topic echo /odom --field twist.twist

header.frame_id는 일반적으로 odom, child_frame_id는 base_link 계열이어야 하지만 Robot 구성에 따라 다릅니다. 값만 보지 말고 Frame과 Timestamp를 함께 확인합니다.

16. 실제 이동 Robot에서 주요 Topic이 흐르는 방법

LiDAR Driver ── /scan ──────────────┐
Wheel Driver ── /wheel_states ──┐   │
IMU Driver ──── /imu/data ──────┤   │
                                 v   v
                         State Estimator / Localization
                                 │
                                 ├── /odom
                                 └── TF: odom → base_link

Goal → Nav2 → /cmd_vel_nav ─┐
Joystick → /cmd_vel_joy ────┼→ Velocity Mux → /cmd_vel_raw
Remote UI → /cmd_vel_web ───┘                    │
                                                 v
                                  Safety Filter + Watchdog
                                                 │ /cmd_vel
                                                 v
                                        Base Controller → Motor

`/scan`

LiDAR 거리 배열입니다. Navigation Costmap, SLAM, Collision Monitor와 Visualization이 동시에 구독할 수 있습니다. 일반적으로 Sensor QoS와 정확한 header.stamp, header.frame_id가 중요합니다.

`/odom`

연속적인 Local Pose와 Twist입니다. Wheel Odometry나 State Estimator가 발행하고 Navigation과 Monitoring이 사용합니다. 장기 Drift할 수 있으며 map Frame의 전역 위치와 혼동하면 안 됩니다.

`/cmd_vel`

Robot Base에 전달할 속도 명령입니다. 여러 Source가 같은 Topic에 동시에 발행하게 두기보다 Twist Multiplexer나 Arbitration Node에서 우선순위를 정하는 것이 안전합니다.

`/joint_states`

Joint Position, Velocity와 Effort를 전달합니다. robot_state_publisher, RViz와 Controller가 사용합니다.

Battery와 Diagnostic Topic

sensor_msgs/msg/BatteryState, /diagnostics 등을 사용해 UI와 Safety Supervisor가 상태를 감시합니다. 단, 고속 Hardware Protection을 느린 ROS Topic만으로 구현해서는 안 됩니다.

17. 여러 Publisher가 한 Topic에 있을 때

Topic은 Many-to-many를 허용합니다. 따라서 두 Publisher가 /cmd_vel에 발행하면 Subscriber는 두 Source의 Message를 모두 받습니다. “마지막 Message가 이긴다”는 명시적인 Arbitration 규칙이 아니라 도착 순서에 따라 명령이 섞이는 상태입니다.

ros2 topic info /cmd_vel --verbose

Publisher count가 예상보다 많으면 Node 이름을 확인합니다.

권장 구조:

Navigation ─┐
Joystick ───┼→ Velocity Mux → Safety Filter → Base Driver
Remote UI ──┘

Mux에서 Source Priority, Timeout과 Lockout을 정의합니다. Emergency Stop은 단순히 가장 높은 우선순위의 0 속도 Topic 하나로만 구현하지 말고 Hardware Safety Chain과 연결합니다.

18. Topic을 Bag으로 기록하고 재생하기

선택 Topic 기록

ros2 bag record -o topic_test \
  /scan /odom /cmd_vel_raw /cmd_vel /tf /tf_static

모든 Topic 기록

ros2 bag record -a

모든 Image와 Point Cloud를 기록하면 Disk와 CPU가 포화될 수 있으므로 필요한 Topic과 Storage 용량을 계산합니다.

기록 정보 확인

ros2 bag info topic_test

재생

ros2 bag play topic_test

실제 Motor Driver가 연결된 상태에서 과거 /cmd_vel까지 재생하면 Robot이 움직일 수 있습니다. 분석용 Bag에서는 Command Topic을 제외하거나, Driver를 끄고 Remap하여 안전한 Topic으로 재생합니다.

ros2 bag play topic_test --remap /cmd_vel:=/recorded_cmd_vel

Bag Replay는 Sensor Algorithm 회귀 Test에 매우 유용하지만 Parameter, Software Version, Calibration과 외부 Service 상태를 자동으로 모두 기록하지 않습니다.

19. Topic 진단 절차

Topic이 안 보이거나 값이 이상할 때 다음 순서로 확인합니다.

1단계 — 환경

printenv ROS_DISTRO
echo "$ROS_DOMAIN_ID"
echo "$RMW_IMPLEMENTATION"

2단계 — Node와 Topic 존재

ros2 node list
ros2 topic list -t
ros2 node info /expected_node

3단계 — Type과 Endpoint

ros2 topic type /scan
ros2 interface show sensor_msgs/msg/LaserScan
ros2 topic info /scan --verbose

4단계 — 실제 Data

ros2 topic echo /scan --once --qos-reliability best_effort
ros2 topic hz /scan
ros2 topic bw /scan

5단계 — 이름 변경

ros2 node info /node_name
rqt_graph

Launch의 Namespace와 Remap을 확인합니다.

6단계 — Network와 Discovery

같은 Domain, Multicast, Firewall, VPN, Container Network와 DDS 설정을 확인합니다. 같은 Computer에서 먼저 시험해 Application 문제와 Network 문제를 분리합니다.

7단계 — Callback 지연

발행 주기는 정상인데 Application 반응이 늦으면 Callback 시작·종료 시간, Queue Depth, CPU와 Data Age를 측정합니다.

20. 증상별 빠른 점검표

증상	먼저 실행할 명령	주요 원인
Topic 자체가 없음	`ros2 topic list -t`	Node 미실행, 이름·Namespace, Discovery
Topic은 있는데 값 없음	`ros2 topic info ... --verbose`	Publisher 0, Type·QoS 불일치
CLI Echo만 값 없음	Best Effort Option으로 Echo	CLI Subscriber QoS 불일치
주기가 낮음	`ros2 topic hz`	Publisher 부하, Network 유실, CLI 수신 부하
Robot 반응이 늦음	Stamp Age와 Callback 시간	깊은 Queue, Blocking Callback
Robot이 떨림	Publisher count 확인	여러 Command Publisher 충돌
종료 후 계속 움직임	Driver Timeout 시험	Deadman·Hardware Watchdog 없음
Image로 Network 포화	`ros2 topic bw`	해상도·Frame Rate·압축·QoS 부적합
다른 Robot Data가 섞임	Namespace·Domain 확인	이름 충돌, Domain 공유

21. Topic 설계 원칙

이름보다 먼저 Message 의미, 단위, Frame과 Timestamp를 정의합니다.
Node Code에는 가능한 상대 Topic 이름을 사용하고 배포 시 Namespace·Remap합니다.
Sensor, Command, State와 Event의 성격에 맞게 QoS와 Depth를 선택합니다.
최신 값이 중요한 Data에 큰 Queue를 두지 않습니다.
Callback을 짧게 유지하고 처리 시간과 Data Age를 측정합니다.
여러 Command Source는 직접 경쟁시키지 말고 명시적인 Arbitration을 둡니다.
Command Subscriber에는 Timeout과 안전한 기본 상태를 둡니다.
Header가 있는 Data는 Frame과 Timestamp를 검증합니다.
CLI로 Publisher와 Subscriber를 각각 대체해 문제 구간을 분리합니다.
ROS Topic과 별개로 Hardware Safety와 Watchdog을 유지합니다.

22. 전체 실습 Check List

[ ] /practice/chatter를 CLI로 한 번과 일정 주기로 발행했다.
[ ] list -t, type, find, interface show로 Type을 조사했다.
[ ] info --verbose에서 Publisher·Subscriber와 QoS를 확인했다.
[ ] echo, hz, bw의 측정 의미를 설명할 수 있다.
[ ] topic_practice Package를 Build하고 Talker·Listener를 실행했다.
[ ] 두 Listener가 같은 Message를 각각 받는 것을 확인했다.
[ ] Remap과 Namespace로 최종 Topic 이름을 바꿨다.
[ ] Sensor QoS로 /scan을 구독했다.
[ ] cmd_vel_filter의 제한, 장애물 차단과 Deadman을 시험했다.
[ ] /odom의 Frame, Timestamp, Pose와 Twist를 확인했다.
[ ] Bag에서 Command Topic을 안전하게 Remap해 재생했다.
[ ] Publisher가 끊겼을 때 실제 Base Driver가 Timeout 정지하는지 검증했다.

핵심 정리

Topic은 Robot의 연속 Data를 전달하는 비동기 Publish·Subscribe 통신입니다.
연결에는 Topic 이름, Message Type과 호환 QoS가 필요합니다.
Publisher와 Subscriber는 서로를 모르며 상대가 없어도 오류 없이 동작합니다.
ros2 topic info --verbose, echo, hz, bw는 연결·값·주기·대역폭 진단의 핵심입니다.
Queue를 키우면 유실 대신 지연과 오래된 Data 처리가 생길 수 있습니다.
Callback은 짧게 유지하고 일정한 출력은 Timer에서 발행하는 것이 안전합니다.
Namespace와 Remap은 같은 Code를 여러 Robot과 System 구성에서 재사용하게 합니다.
실제 Robot의 /scan, /odom, /cmd_vel은 Frame, Timestamp, QoS와 안전 Timeout을 함께 설계해야 합니다.
여러 속도 Publisher는 Mux로 중재하고 Motor Driver에도 독립 Watchdog을 둡니다.
Bag Replay는 강력한 Test 도구지만 실제 Command 재생은 Robot을 움직일 수 있으므로 격리·Remap해야 합니다.

용어 전에 읽는 공식 참고자료

ROBOT GLOSSARY

용어 정리

전체 용어 찾아보기 →

Publish-Subscribe발행·구독 통신: 발행자가 Topic에 Message를 보내고 하나 이상의 구독자가 비동기로 받아 처리하는 느슨하게 결합된 통신 방식입니다.
Topic NameTopic 이름: ROS Graph에서 Message Stream을 식별하며 Namespace와 Remapping의 영향을 받는 논리적 통신 경로입니다.
Message TypeMessage 자료형: Topic에서 교환하는 Data Field와 Type을 정의하는 Interface 계약으로 Publisher와 Subscriber가 같아야 합니다.
Subscription구독 객체: 특정 Topic과 Type의 Message 도착을 기다리고 등록된 Callback에 전달하는 Subscriber 측 ROS Entity입니다.
Callback이벤트 처리 함수: Message, Timer나 Service Event가 준비되었을 때 Executor가 호출하는 사용자 함수입니다.
Queue Depth보관 Message 개수: Keep Last History에서 아직 처리하지 못했거나 전달할 최근 Message를 몇 개 유지할지 정하는 QoS 값입니다.
Best Effort최선형 전달: 재전송보다 최신 Data와 낮은 지연을 우선해 일부 Message 유실을 허용하는 Reliability 정책입니다.
Reliable신뢰형 전달: Middleware가 정해진 범위에서 확인과 재전송을 이용해 Message 전달 신뢰도를 높이는 QoS 정책입니다.
Sensor Data QoSSensor용 QoS Profile: 오래된 Sample의 완전한 전달보다 최신 측정의 빠른 수신을 우선하도록 Best Effort와 얕은 Queue를 사용하는 Profile입니다.
Remapping이름 재지정: Node Source를 바꾸지 않고 실행 또는 Launch 시 Topic, Service나 Node 이름을 다른 이름에 연결하는 기능입니다.
Namespace이름 공간: Node와 Topic 이름 앞에 계층적 Prefix를 붙여 여러 Robot이나 Subsystem의 Graph 이름 충돌을 줄이는 구조입니다.
Topic IntrospectionTopic 실행 상태 조사: CLI와 Graph Tool로 Topic Type, Endpoint, QoS, 값, 주기와 대역폭을 실행 중 확인하는 작업입니다.
Message RateMessage 수신 주기: Subscriber가 단위 시간에 실제로 받은 Message 수로 보통 Hz 단위로 측정합니다.
Bandwidth통신 대역폭: Topic Subscriber가 단위 시간에 수신한 Data 양으로 Message 크기와 주기의 영향을 함께 받습니다.
Deadman Timeout명령 단절 자동 정지: 정해진 시간 동안 새 Command가 없으면 마지막 명령을 폐기하고 안전한 정지 상태로 전환하는 Watchdog 규칙입니다.
Velocity Multiplexer속도 명령 중재기: Navigation, Joystick과 원격 UI 등 여러 속도 Source의 Priority와 Timeout을 적용해 하나의 명령을 선택하는 Component입니다.
LaserScan2D LiDAR Scan Message: 각도 범위와 증가량, 거리 배열, 측정 범위와 Timestamp·Frame을 담는 ROS Sensor Message입니다.
Odometry오도메트리: Local Frame에서 추정한 Robot Pose와 Twist 및 그 불확실성을 시간과 Frame 정보와 함께 표현하는 Message입니다.
Data Age측정 Data 나이: 현재 ROS Time과 Message Header의 측정 Timestamp 차이로 Queue·Network·처리 지연을 포함한 Data의 오래됨을 나타냅니다.
Command Watchdog명령 감시 장치: 새 제어 명령의 도착을 감시하고 Timeout 시 하위 Controller나 Hardware를 안전 상태로 전환하는 보호 기능입니다.

연습 문제

Topic이 Service보다 Sensor Data에 적합한 이유를 설명하세요.
Publisher와 Subscriber가 연결되기 위한 세 조건은 무엇인가요?
Publisher count가 1인데 Callback이 실행되지 않을 때 어떤 항목을 확인해야 하나요?
ros2 topic hz 결과가 Publisher 내부 주기와 다를 수 있는 이유는 무엇인가요?
ros2 topic info --verbose에서 확인할 수 있는 정보를 네 가지 쓰세요.
상대 Topic 이름이 Multi-robot System에서 유리한 이유는 무엇인가요?
Reliable Subscriber가 Best Effort LiDAR Publisher를 받지 못할 때 어떻게 진단하고 수정하나요?
Queue Depth를 1000으로 크게 설정하는 것이 제어 Topic에 위험한 이유는 무엇인가요?
Subscription Callback에서 무거운 Image 처리를 하면 /cmd_vel 처리에 어떤 영향이 생기나요?
cmd_vel_filter에서 구독 Callback과 20 Hz 출력 Timer를 분리한 이유는 무엇인가요?
Deadman Timeout과 Emergency Stop의 역할 차이는 무엇인가요?
여러 Node가 /cmd_vel에 직접 발행할 때 생기는 문제와 해결 구조를 설명하세요.
/odom Message에서 값 외에 Frame과 Timestamp를 확인해야 하는 이유는 무엇인가요?
실제 Robot이 연결된 상태에서 /cmd_vel이 든 Bag을 재생할 때 필요한 안전 조치는 무엇인가요?
Topic이 안 보이는 문제를 Application, QoS, Network 순서로 분리해 진단하는 절차를 설명하세요.