본문 바로가기

어디부터 읽을까요?

드론·제어

제어 원리에서 ROS 2·PX4 비행 구현까지 연결합니다.

로봇·Physical AI

로봇이 보고, 배우고, 행동하는 데 필요한 개념을 읽습니다.

실용 개발

바로 적용할 작은 예제로 개발 중 막힌 문제를 해결합니다.

학습 기록 2편 — 항공 원리
로봇 개발 실전

ROS 2에서 토픽은 보이는데 데이터가 안 들어오는 이유

반응형

ros2 topic list에 이름이 보여도 Publisher와 Subscriber가 메시지를 주고받는다는 뜻은 아닙니다. 먼저 토픽 타입과 실제 Publisher, QoS를 확인하세요. 센서의 Best Effort 발행자에 Reliable 수신자가 붙으면 이름은 보여도 연결되지 않을 수 있습니다.

 

개념 설명용 도해: 발행자 → Reliable 수신 → Best Effort 수신입니다.

1. 확인 순서를 좁히기

ros2 topic info /대상토픽 -v로 Publisher 수·노드·메시지 타입·QoS를 확인합니다. Publisher가 없다면 프로그램 실행이나 토픽 이름부터 보세요. Publisher가 있는데 수신이 없다면 Reliability와 Durability를 비교하세요. 그다음 시계·필터·콜백 문제를 살펴보면 원인을 한꺼번에 섞지 않을 수 있습니다.

2. 요청한 품질을 제공할 수 있어야 연결됩니다

Reliable Subscriber는 Reliable 전송을 요청합니다. Best Effort Publisher는 그 요청을 만족하지 못합니다. 반대로 Reliable Publisher와 Best Effort Subscriber는 Reliability 정책 기준으로 연결할 수 있습니다. 모든 정책이 함께 호환되어야 하므로 Reliability 하나만 맞췄다고 문제를 다 해결한 것으로 보지는 않습니다.

3. 도구 수신도 같은 조건으로 비교하기

센서가 Best Effort로 발행한다면 ros2 topic echo /대상토픽 --qos-reliability best_effort로 확인하세요. 기본 echo가 우연히 연결되는 경우도 있으므로 명령의 기본값을 추측하지 말고 실제 endpoint 설정을 함께 봅니다. 아래 예제는 한 Publisher에 두 Subscriber를 붙여 차이를 직접 보여줍니다. ROS 2 Jazzy에서 실행하며 실제 센서나 PX4 연결은 필요하지 않습니다.

직접 실행하기

다음 코드를 example.py로 저장한 뒤 ROS 2 Jazzy 환경에서 python3 example.py를 실행하세요.

import os
import time
os.environ.setdefault("ROS_DOMAIN_ID", "188")
import rclpy
from rclpy.node import Node
from rclpy.qos import QoSProfile, ReliabilityPolicy
from std_msgs.msg import String

rclpy.init()
node = Node("qos_receive_example")
counts = {"reliable": 0, "best_effort": 0}
def callback(name):
    def receive(msg):
        counts[name] += 1
    return receive
best = QoSProfile(depth=10, reliability=ReliabilityPolicy.BEST_EFFORT)
reliable = QoSProfile(depth=10, reliability=ReliabilityPolicy.RELIABLE)
pub = node.create_publisher(String, "/qos_receive_example", best)
sub_a = node.create_subscription(String, "/qos_receive_example", callback("reliable"), reliable)
sub_b = node.create_subscription(String, "/qos_receive_example", callback("best_effort"), best)
timer = node.create_timer(0.05, lambda: pub.publish(String(data="hello")))
end = time.monotonic() + 3
while time.monotonic() < end:
    rclpy.spin_once(node, timeout_sec=0.05)
print("Reliable 수신:", counts["reliable"])
print("Best Effort 수신:", counts["best_effort"])
assert counts["reliable"] == 0 and counts["best_effort"] > 0
node.destroy_node()
rclpy.shutdown()

실행 결과:

Reliable 수신: 0
Best Effort 수신: 59

터미널에는 incompatible QoS 경고도 표시됩니다. Best Effort 수신 개수는 발견 시간과 실행 환경에 따라 달라집니다. 여기서는 Reliable이 0이고 Best Effort가 양수인지 확인했습니다. 다음 편에서는 네 가지 Reliability 조합과 Durability를 표로 정리하겠습니다.

이어서 읽기

참고 자료

반응형