Subscriber trong ROS2 là gì?

Nếu Publisher là bên "nói", Subscriber là bên "nghe" — một node muốn nhận dữ liệu từ một topic (dữ liệu LiDAR, lệnh điều khiển, trạng thái pin...) sẽ tạo một Subscriber gắn với topic và kiểu message tương ứng.

Khái niệm chính

Điểm khác biệt quan trọng nhất của Subscriber so với việc "hỏi dữ liệu thủ công": nó hoạt động theo mô hình hướng sự kiện (event-driven). Khi tạo subscriber, node đăng ký kèm một hàm callback — hàm này tự động được gọi mỗi khi có message mới tới trên topic đó, node không cần chủ động kiểm tra ("có dữ liệu chưa? có dữ liệu chưa?") theo kiểu polling tốn tài nguyên.

Việc callback được gọi đúng lúc là nhiệm vụ của executor — cơ chế bên trong rclpy.spin() liên tục theo dõi mọi subscriber/timer/service của node và gọi đúng hàm xử lý tương ứng khi sự kiện xảy ra.

Tóm lại: Không cần vòng lặp while True: kiểm_tra_dữ_liệu() thủ công — chỉ cần định nghĩa callback đúng logic xử lý, rclpy.spin() sẽ tự lo phần "khi nào gọi nó".

Nguyên lý hoạt động

Subscriber tối thiểu, in ra vận tốc mỗi khi nhận được message mới trên /cmd_vel:

import rclpy
from rclpy.node import Node
from geometry_msgs.msg import Twist

class VelocitySubscriber(Node):
    def __init__(self):
        super().__init__('velocity_subscriber')
        self.subscription = self.create_subscription(
            Twist, '/cmd_vel', self.listener_callback, 10)

    def listener_callback(self, msg):
        # Hàm này TỰ ĐỘNG được gọi mỗi khi có message mới — không cần vòng lặp thủ công
        self.get_logger().info(f'Nhận được: linear.x = {msg.linear.x}')

def main():
    rclpy.init()
    node = VelocitySubscriber()
    rclpy.spin(node)   # giữ node sống, executor gọi listener_callback khi có dữ liệu
    rclpy.shutdown()

Message mới tới trên /cmd_vel
         ↓
   executor phát hiện
         ↓
   gọi listener_callback(msg)
         ↓
   code xử lý bên trong callback chạy

Một lưu ý quan trọng khi viết callback: không nên đặt code chạy lâu (vòng lặp nặng, sleep() dài) bên trong callback — vì executor thường xử lý callback tuần tự trên cùng một luồng, một callback chạy lâu sẽ chặn luôn việc xử lý các sự kiện khác của node (timer, subscriber khác) cho tới khi nó chạy xong.

Kiến thức kỹ thuật

Công thức

Để callback không bị "dồn ứ" (message tới nhanh hơn tốc độ xử lý), thời gian xử lý mỗi callback nên nhỏ hơn nhiều so với chu kỳ giữa 2 message liên tiếp:

Thời gian xử lý callback phải << 1 / tần suất publish của topic đó

VD: topic publish ở 50Hz (chu kỳ 20ms)
Callback nên xử lý xong trong dưới ~5-10ms để có dư an toàn,
không chiếm gần hết 20ms — nếu không, dù chưa "quá hạn" hẳn, hệ thống
gần như không còn thời gian dự phòng cho jitter hoặc callback khác chen vào

Thông số

Thời gian xử lý callback thực tế khác nhau rất lớn tuỳ loại tác vụ — tác vụ nặng cần chuyển ra khỏi callback chính hoặc dùng threading/callback group riêng (xem chuyên mục ROS2 nâng cao):

Loại xử lý trong callback	Thời gian điển hình
Ghi log, cập nhật 1 biến đơn giản	Dưới 1ms
Tính toán số học nhẹ (VD: chuyển đổi đơn vị)	Vài ms
Xử lý ảnh (OpenCV cơ bản)	Vài chục ms
Suy luận AI/deep learning trên ảnh	Có thể 100ms+

Ứng dụng thực tế

Subscriber cho topic an toàn như /emergency_stop là ví dụ điển hình của callback cần cực kỳ nhẹ và nhanh — logic bên trong chỉ nên set 1 cờ trạng thái (self.estop_triggered = True) rồi return ngay, để độ trễ phản ứng luôn ở mức tối thiểu; toàn bộ xử lý phức tạp hơn (dừng động cơ, ghi log sự kiện, thông báo) nên đặt ở nơi khác đọc cờ này định kỳ, không đặt trực tiếp trong callback an toàn — đúng nguyên tắc "không block trong callback" đã nói ở phần Nguyên lý hoạt động, nhưng ở đây hậu quả của việc vi phạm nghiêm trọng hơn nhiều vì liên quan trực tiếp tới an toàn vận hành robot.

Ví dụ thực tế

# Subscriber dạng lambda ngắn gọn cho tác vụ đơn giản
self.create_subscription(
    Bool, '/emergency_stop',
    lambda msg: setattr(self, 'estop_triggered', msg.data),
    10)

$ ros2 topic pub /emergency_stop std_msgs/msg/Bool "{data: true}" --once

Output (log của node subscriber, nếu có ghi log trong callback):
[INFO] [velocity_subscriber]: Nhận được tín hiệu dừng khẩn cấp

Callback dạng lambda 1 dòng minh hoạ đúng nguyên tắc "càng nhẹ càng tốt" cho dữ liệu an toàn — không có logic phức tạp nào có thể làm chậm phản ứng, chỉ đơn giản gán giá trị vào 1 biến trạng thái.

Lỗi thường gặp

Đặt code chạy lâu bên trong callback — như đã nói ở phần Nguyên lý hoạt động, một callback chậm chặn toàn bộ việc xử lý sự kiện khác của node (executor mặc định đơn luồng) — nếu node "phản ứng chậm" một cách khó hiểu, kiểm tra callback nào đang mất nhiều thời gian trước tiên.
Tạo subscriber nhưng quên rclpy.spin(node) — giống lỗi đã nói ở bài Node, không gọi spin() thì callback không bao giờ được thực thi dù subscriber đã tạo đúng.
QoS không khớp với publisher khiến không nhận được message nào — subscriber code hoàn toàn đúng, nhưng nếu chính sách QoS (reliability, durability) không tương thích với publisher, kết nối không thiết lập được — không có lỗi báo rõ ràng, chỉ đơn giản callback không bao giờ được gọi.
Truy cập field không tồn tại trong message do nhầm phiên bản kiểu dữ liệu — nếu định nghĩa message tuỳ chỉnh thay đổi (thêm/bớt field) nhưng subscriber chưa build lại theo phiên bản mới nhất, truy cập field cũ/mới không khớp có thể gây lỗi runtime — luôn rebuild toàn bộ workspace sau khi sửa định nghĩa message dùng chung.

Kết luận

Subscriber hoạt động theo mô hình hướng sự kiện, giải phóng lập trình viên khỏi việc tự viết vòng lặp polling — nhưng đổi lại phải cẩn trọng giữ callback càng nhẹ càng tốt, đặc biệt với dữ liệu liên quan an toàn. Kết hợp Publisher và Subscriber đã học đủ để xây dựng giao tiếp topic hoàn chỉnh giữa 2 node. Bài tiếp theo trong chuyên mục nói về Service — mô hình giao tiếp yêu cầu-phản hồi, khác hẳn kiểu fire-and-forget của topic.