토픽 발행과 구독 - 센서 값을 흘려보내기

이 장에서 배우는 것

앞 장에서는 rclpy 로 노드 하나를 만들고 실행하는 법을 다뤘다. 노드 혼자서는 아무 일도 하지 않는다. 두리가 앞에 장애물이 있는지 판단하려면 거리 센서 값을 어딘가로 흘려보내야 하고, 그 값을 안전 정지 로직이나 화면 표시 기능이 받아서 써야 한다. 이 장에서는 ROS 2 가 데이터를 흘려보내는 기본 통로인 토픽(topic)을 다룬다. 퍼블리셔(publisher)가 값을 내보내고 구독자(subscriber)가 그 값을 받는 구조, 그리고 이 둘을 연결하는 큐 크기와 명령행 도구까지 살펴본다.

• 퍼블리셔와 구독자가 토픽을 통해 어떻게 데이터를 주고받는지 설명할 수 있다
• std_msgs 의 기본 메시지 타입으로 퍼블리셔와 구독자 노드를 작성할 수 있다
• 큐 크기(queue size)가 무엇을 조절하는 값인지 이해한다
• ros2 topic echo/hz/pub 명령으로 실행 중인 토픽을 들여다볼 수 있다

문제 상황

두리 소프트웨어를 처음 짤 때 흔히 저지르는 실수는, 센서 값을 읽는 코드 안에 그 값을 쓰는 코드까지 같이 집어넣는 것이다. 예를 들어 거리 센서를 읽는 함수 안에서 곧바로 "너무 가까우면 정지"라는 판단 함수를 호출하고, 로그를 남기는 함수도 그 자리에서 호출하는 식이다. 처음에는 문제가 없어 보이지만, 나중에 거리 값을 화면에 표시하는 기능이나 주행 기록을 저장하는 기능을 추가하려면 센서 읽는 코드를 또 고쳐야 한다. 센서 하나를 담당하는 코드가 그 값을 쓰는 모든 기능을 알아야 하는 구조는 기능이 늘어날수록 점점 더 손대기 어려워진다.

ROS 2 는 이 문제를 퍼블리셔-구독자 구조로 푼다. 센서 값을 만드는 노드는 "이 이름의 토픽에 값을 흘려보낸다"는 것만 알면 되고, 그 값을 누가 몇 명이 받는지는 신경 쓰지 않는다. 값을 쓰는 노드는 "이 이름의 토픽을 구독한다"고만 선언하면 값이 들어올 때마다 콜백이 불린다. 두 노드는 서로의 존재를 몰라도 된다.

퍼블리셔, 구독자, 토픽

토픽은 이름과 메시지 타입으로 정의되는 통로다. 예를 들어 /duri/front_distance 라는 이름에 std_msgs/msg/Float32 타입의 메시지가 흐르기로 정했다면, 이 토픽에 발행하는 쪽과 구독하는 쪽은 이름과 타입이 정확히 같아야 연결된다. 이름이나 타입 중 하나라도 다르면 두 노드는 같은 주제를 말하는 것처럼 보여도 실제로는 전혀 다른 토픽을 쓰는 것이 되어 데이터가 흐르지 않는다. 이때 ROS 2 는 오류를 내지 않는다. 그냥 조용히 아무 일도 일어나지 않을 뿐이다. 이 특징 때문에 토픽 이름은 노드 코드 여기저기에 문자열로 흩어놓기보다 한 곳에 상수로 모아두는 습관이 유용하다.

퍼블리셔는 create_publisher(메시지타입, 토픽이름, 큐크기) 로 만들고, 구독자는 create_subscription(메시지타입, 토픽이름, 콜백함수, 큐크기) 로 만든다. 구독자 쪽 콜백은 메시지가 도착할 때마다 rclpy 실행기(executor)가 대신 호출해준다. 노드 코드에서 직접 함수를 호출하는 구조가 아니라, 메시지가 오면 등록해둔 콜백이 실행되는 구조라는 점이 핵심이다. 아래 그림은 퍼블리셔 하나가 내보낸 값이 토픽을 거쳐 구독 중인 노드와 명령행 도구 양쪽에 동시에 전달되는 흐름을 보여준다.

std_msgs 와 큐 크기

std_msgs 패키지는 Float32, Int32, String, Bool 처럼 값 하나만 담는 기본 메시지 타입을 모아둔 패키지다. 실제 로봇 프로젝트에서는 센서마다 의미 있는 필드를 여러 개 담은 사용자 정의 메시지를 쓰는 경우가 많지만, 그 내용은 다음 장에서 다룬다. 이 장에서는 거리 값 하나만 흘려보내면 충분하므로 std_msgs/msg/Float32 를 그대로 쓴다.

create_publisher 와 create_subscription 의 마지막 정수 인자가 큐 크기다. 이 값은 구독자가 콜백을 처리하는 속도가 발행 속도를 따라가지 못할 때, 처리를 기다리는 메시지를 몇 개까지 쌓아둘지 정한다. 큐가 가득 찬 상태에서 새 메시지가 도착하면 가장 오래된 메시지부터 버려진다. 큐 크기를 1로 두면 최신 값만 남기고 나머지는 버리는 셈이라 짧은 순간의 변화를 놓치기 쉽고, 반대로 지나치게 크게 두면 처리 지연이 쌓여 구독자가 받는 값이 실제 상황보다 뒤처질 수 있다. 이 책에서는 대부분 10을 기본값으로 쓴다. 실제 ROS 2 에는 큐 크기 말고도 신뢰성이나 지속성을 조절하는 QoS(품질 정책) 설정이 더 있는데, 자세한 내용은 공식 문서의 QoS 설명을 참고한다.

ros2 topic 명령으로 흐름 확인하기

노드를 실행한 뒤에는 ros2 topic 하위 명령으로 실제로 데이터가 흐르는지 눈으로 확인할 수 있다. 코드를 고치지 않고도 토픽 이름이 맞는지, 값이 얼마나 자주 오는지, 특정 값을 한 번 밀어 넣어 구독자를 테스트할 수 있는지를 바로 알 수 있어서 디버깅 단계에서 자주 쓴다.

ros2 topic 하위 명령이 하는 일

명령	하는 일	예시
list	현재 떠 있는 토픽 이름을 모두 나열한다	ros2 topic list
echo	토픽에 흐르는 메시지를 화면에 그대로 찍는다	ros2 topic echo /duri/front_distance
hz	메시지가 얼마나 자주 오는지 초당 횟수로 잰다	ros2 topic hz /duri/front_distance
pub	터미널에서 직접 메시지를 발행해 구독자를 테스트한다	ros2 topic pub /duri/front_distance std_msgs/msg/Float32 "{data: 0.15}" --once

큐 크기가 실제로 어떻게 동작하는지는 ROS 2 를 띄우지 않고도 확인할 수 있다. 아래 그림은 큐 크기가 3인 토픽에 새 메시지가 도착했을 때, 가장 오래된 값이 밀려나는 과정을 보여준다.

완성 코드

duri_distance_publisher.py

import random

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


class DistancePublisher(Node):

    def __init__(self):
        super().__init__('duri_distance_publisher')
        self.publisher_ = self.create_publisher(Float32, 'duri/front_distance', 10)
        self.timer = self.create_timer(0.1, self.publish_distance)
        self.get_logger().info('전방 거리 센서 퍼블리셔를 시작한다')

    def publish_distance(self):
        msg = Float32()
        msg.data = round(random.uniform(0.1, 2.0), 2)
        self.publisher_.publish(msg)


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


if __name__ == '__main__':
    main()

duri_distance_subscriber.py

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

SAFE_DISTANCE = 0.3


class DistanceSubscriber(Node):

    def __init__(self):
        super().__init__('duri_distance_subscriber')
        self.subscription = self.create_subscription(
            Float32,
            'duri/front_distance',
            self.listener_callback,
            10)

    def listener_callback(self, msg):
        if msg.data < SAFE_DISTANCE:
            self.get_logger().warn(f'전방 {msg.data} m, 장애물 근접')
        else:
            self.get_logger().info(f'전방 거리: {msg.data} m')


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


if __name__ == '__main__':
    main()

topic_sim.py (ROS 없이 확인하는 메시지 흐름)

from collections import deque


class Topic:
    def __init__(self, name, queue_size):
        self.name = name
        self.queue = deque(maxlen=queue_size)
        self.subscribers = []

    def subscribe(self, callback):
        self.subscribers.append(callback)

    def publish(self, message):
        if len(self.queue) == self.queue.maxlen:
            print(f'[{self.name}] 큐가 가득 차 가장 오래된 값을 버렸다: {self.queue[0]:.2f} m')
        self.queue.append(message)

    def spin_once(self):
        while self.queue:
            message = self.queue.popleft()
            for callback in self.subscribers:
                callback(message)


def make_logger(name):
    def callback(message):
        print(f'[{name}] 수신: {message:.2f} m')
    return callback


def main():
    front_distance = Topic('front_distance', queue_size=3)
    front_distance.subscribe(make_logger('safety_monitor'))

    readings = [1.20, 0.95, 0.40, 0.25, 0.60]
    for value in readings:
        front_distance.publish(value)

    front_distance.spin_once()


if __name__ == '__main__':
    main()

줄별 해설

duri_distance_publisher.py 는 앞 장에서 다룬 노드 뼈대에 퍼블리셔 하나를 더한 것이다. create_publisher(Float32, 'duri/front_distance', 10) 는 이 노드가 duri/front_distance 라는 이름으로 Float32 타입 메시지를 내보내겠다고 선언하고, 큐 크기는 10으로 둔다. create_timer(0.1, self.publish_distance) 는 0.1초마다 publish_distance 를 호출해 초당 10회 값을 흘려보낸다. 실제 두리에는 초음파 센서 드라이버가 값을 채워주겠지만, 이 장에서는 random.uniform 으로 0.1~2.0 사이의 값을 흉내 낸다.

duri_distance_subscriber.py 는 같은 이름과 타입으로 create_subscription 을 호출한다. 이름과 타입이 퍼블리셔 쪽과 정확히 같아야 메시지가 연결된다. 메시지가 도착할 때마다 listener_callback 이 불리고, 거리가 SAFE_DISTANCE(0.3m)보다 가까우면 경고를 남긴다. 콜백 안에는 조건 분기와 로그 출력만 있을 뿐 무거운 계산이 없다는 점에 주목한다. 이유는 뒤에서 다룬다.

topic_sim.py 는 ROS 2 없이 퍼블리셔-구독자 구조와 큐 크기의 동작을 확인하는 순수 파이썬 예제다. Topic 클래스는 subscribers 리스트에 콜백을 등록해두고, publish 는 값을 큐에 넣기만 한다. 큐는 deque(maxlen=queue_size) 로 만들어서, 가득 찬 상태에서 값이 더 들어오면 가장 오래된 값이 자동으로 밀려난다. publish 안의 if len(self.queue) == self.queue.maxlen: 은 밀려나기 직전에 어떤 값이 버려지는지 미리 확인해서 출력하는 부분이다. spin_once 는 실제 rclpy 의 spin 을 흉내 낸 것으로, 큐에 쌓인 메시지를 순서대로 꺼내 구독자 콜백을 호출한다.

실행 결과

두 노드 파일은 앞 장에서 만든 duri_bringup 패키지 안에 추가하고, setup.py 의 entry_points 에 콘솔 스크립트로 등록했다고 가정한다. 터미널 두 개를 열어 각각 실행한다.

$ ros2 run duri_bringup duri_distance_publisher
[INFO] [1758000000.123456] [duri_distance_publisher]: 전방 거리 센서 퍼블리셔를 시작한다

$ ros2 run duri_bringup duri_distance_subscriber
[INFO] [1758000000.234567] [duri_distance_subscriber]: 전방 거리: 1.47 m
[WARN] [1758000000.334567] [duri_distance_subscriber]: 전방 0.22 m, 장애물 근접
[INFO] [1758000000.434567] [duri_distance_subscriber]: 전방 거리: 0.88 m

값은 무작위로 만들어지므로 화면에 나오는 숫자와 시각은 실행할 때마다 다르다. 세 번째 터미널에서 토픽을 직접 들여다볼 수도 있다.

$ ros2 topic echo /duri/front_distance
data: 1.47
---
data: 0.22
---
data: 0.88
---

$ ros2 topic hz /duri/front_distance
average rate: 10.001
	min: 0.098s max: 0.102s std dev: 0.00091s window: 20

순수 파이썬 보조 예제는 ROS 2 없이 그 자리에서 확인할 수 있다.

$ python3 topic_sim.py
[front_distance] 큐가 가득 차 가장 오래된 값을 버렸다: 1.20 m
[front_distance] 큐가 가득 차 가장 오래된 값을 버렸다: 0.95 m
[safety_monitor] 수신: 0.40 m
[safety_monitor] 수신: 0.25 m
[safety_monitor] 수신: 0.60 m

큐 크기가 3인데 다섯 개의 값을 연달아 발행했으므로 앞의 두 값(1.20, 0.95)은 spin_once 가 불리기 전에 밀려나 버려졌고, 남은 세 값만 구독자에게 전달됐다.

실무에서 자주 틀리는 것

큐 크기를 1로 두고 값이 자주 사라진다고 오해한다

큐 크기 1은 새 값이 오는 순간 이전 값을 무조건 버린다는 뜻이다. 구독자 처리 속도가 조금만 늦어도 중간 값들이 통째로 사라진다.

self.publisher_ = self.create_publisher(Float32, 'duri/front_distance', 1)

self.publisher_ = self.create_publisher(Float32, 'duri/front_distance', 10)

토픽 이름 오타로 구독자가 아무것도 받지 못한다

이름이 한 글자만 달라도 ROS 2 는 오류를 내지 않고 조용히 연결하지 않는다. ros2 topic list 로 양쪽 이름을 비교해보는 습관이 필요하다.

self.subscription = self.create_subscription(
    Float32, 'duri/front_distnace', self.listener_callback, 10)

self.subscription = self.create_subscription(
    Float32, 'duri/front_distance', self.listener_callback, 10)

메시지 타입을 다르게 지정한다

이름이 같아도 타입이 다르면 별개의 토픽으로 취급된다. 퍼블리셔가 Float32 로 내보내는데 구독자가 Float64 로 구독하면 연결되지 않는다.

from std_msgs.msg import Float64
self.subscription = self.create_subscription(
    Float64, 'duri/front_distance', self.listener_callback, 10)

from std_msgs.msg import Float32
self.subscription = self.create_subscription(
    Float32, 'duri/front_distance', self.listener_callback, 10)

콜백 안에서 시간이 오래 걸리는 작업을 한다

구독 콜백은 rclpy 실행기가 순서대로 처리한다. 콜백 하나가 오래 걸리면 그동안 다른 메시지 처리와 타이머가 모두 밀린다. 무거운 계산은 값만 저장해두고 다른 곳에서 처리한다.

def listener_callback(self, msg):
    time.sleep(1.0)
    self.get_logger().info(f'전방 거리: {msg.data} m')

def listener_callback(self, msg):
    self.last_distance = msg.data
    if msg.data < SAFE_DISTANCE:
        self.get_logger().warn(f'전방 {msg.data} m, 장애물 근접')

한눈에 보기

토픽 발행/구독 핵심 개념

용어	의미	이 장 예제에서 값
토픽 이름	퍼블리셔와 구독자를 연결하는 문자열 키	duri/front_distance
메시지 타입	토픽에 흐르는 데이터의 구조	std_msgs/msg/Float32
큐 크기	구독자가 처리를 기다리는 메시지 최대 개수	10 (topic_sim.py 예제는 3)
퍼블리셔	토픽에 값을 내보내는 쪽, 구독자를 모른다	DistancePublisher
구독자	토픽 값이 올 때마다 콜백이 불리는 쪽	DistanceSubscriber

연습 문제

1. 퍼블리셔의 create_timer 주기를 0.1초에서 1.0초로 바꾸면 ros2 topic hz /duri/front_distance 의 average rate 값이 어떻게 달라지는지 설명하라.
2. topic_sim.py 의 Topic('front_distance', queue_size=5) 로 바꾸고 같은 다섯 개의 값을 발행한 뒤 spin_once 를 호출하면 출력이 어떻게 달라지는지 예측하고, 실제로 실행해서 확인하라.
3. 구독자 노드의 토픽 이름을 duri/front_distnace 로 잘못 입력했다고 하자. 노드는 오류 없이 실행된다. 이 상황을 명령행에서 어떻게 진단할 수 있는지 두 가지 방법을 들어 설명하라.
4. 두리에 후방 거리 센서를 추가해서 duri/rear_distance 토픽으로도 값을 흘려보내려 한다. DistancePublisher 클래스를 어떻게 바꿔야 하는지 방향을 설명하라(전체 코드를 다 쓸 필요는 없다).

정답과 해설

1. 퍼블리셔가 초당 1회만 메시지를 내보내므로 average rate 는 10에서 1에 가까운 값으로 줄어든다. 큐 크기는 초당 발행 횟수 자체를 바꾸지 않으므로 큐 설정과는 무관한 문제다.
2. 읽어 들이는 값이 다섯 개인데 큐 크기가 5이므로 큐가 가득 차는 순간이 오지 않는다. 따라서 "버렸다" 메시지는 한 번도 출력되지 않고, spin_once 호출 시 다섯 값이 발행한 순서 그대로(1.20, 0.95, 0.40, 0.25, 0.60) 모두 safety_monitor 로 전달된다.
3. 첫 번째는 ros2 topic list 로 실행 중인 토픽 이름 목록을 보고 퍼블리셔 쪽 이름과 구독자 코드에 적은 이름이 정확히 같은지 눈으로 비교하는 것이다. 두 번째는 ros2 topic echo /duri/front_distance 로 퍼블리셔가 실제로 값을 내보내는지 확인한 뒤, 구독자 노드의 로그에는 아무것도 찍히지 않는다면 이름 불일치를 의심할 수 있다.
4. 새 퍼블리셔를 하나 더 만들면 된다. __init__ 안에 self.rear_publisher_ = self.create_publisher(Float32, 'duri/rear_distance', 10) 를 추가하고, 타이머 콜백 안에서 전방 값을 발행하는 것과 같은 방식으로 후방 값을 만들어 self.rear_publisher_.publish(...) 로 내보내면 된다. 토픽 이름만 다르면 하나의 노드가 여러 토픽에 동시에 발행할 수 있다.