主題（Topic）與 Pub/Sub

2026-07-18
  • ros2
  • topic
  • qos
  • pub-sub
  • custom-message

問題定義

主題（Topic）是 ROS2 裡最常用的通訊機制，寫一個 publisher 加一個 subscriber 看起來很簡單。但實務上真正棘手的部分，往往不是「怎麼發、怎麼收」，而是兩端的 QoS（Quality of Service）設定沒有對齊——這會導致訊息完全收不到，而且不會有任何錯誤訊息提示你發生了什麼事。這一節除了基本的 pub/sub 寫法，會把 QoS 的三個核心維度講清楚，並示範怎麼建立自己的訊息型別。

核心概念說明

主題通訊的本質：多對多、鬆耦合

一個主題可以有多個 publisher、多個 subscriber，彼此互不知道對方的存在（也不需要知道）——這是「發布訂閱」模式跟後面會教的「服務」模式最根本的差異：主題通訊裡沒有「一問一答」的配對關係，任何時候發布的訊息，會廣播給當下所有已訂閱的節點。

QoS 的三個關鍵維度

DDS 帶來的 QoS 設定，決定了訊息傳遞的可靠度與行為。對日常開發影響最大的三個維度：

Reliability（可靠性）

  • RELIABLE：保證訊息送達，遺失會重傳，但可能增加延遲
  • BEST_EFFORT：盡力傳送，不重傳，遺失就算了

Durability（持久性）

  • VOLATILE：只有「訂閱當下之後」發布的訊息才會收到
  • TRANSIENT_LOCAL：Publisher 會保留最近發布的訊息，新加入的 Subscriber 也能收到「加入之前」發布過的內容

History（歷史紀錄深度）

  • KEEP_LAST(depth=N)：只保留最近 N 筆訊息
  • KEEP_ALL：保留所有訊息（受限於系統資源）

關鍵規則：Publisher 與 Subscriber 的 QoS 必須「相容」，才能建立連線。 相容性判斷不是「完全相同」，而是有明確的相容矩陣，其中最常讓人誤踩的一條是：Subscriber 要求 RELIABLE，但 Publisher 只提供 BEST_EFFORT，兩者不相容（反過來——Subscriber 要求 BEST_EFFORT，Publisher 提供 RELIABLE，則是相容的，因為 Subscriber 的要求比較寬鬆）。

實務上常見的搭配：

  • 高頻率感測器資料流（雷射掃描、影像）：BEST_EFFORT + VOLATILE + KEEP_LAST(depth=5)——新資料很快就會覆蓋舊資料，沒必要保證每筆都送達
  • 地圖、機器人狀態這類「新加入者也要看到目前狀態」的資料：RELIABLE + TRANSIENT_LOCAL——晚加入的節點也能拿到最新發布過的內容

實作範例：完整的 Publisher + Subscriber

1. Publisher 節點

檔案位置：~/ros2_ws/src/my_package/my_package/talker.py

python
import rclpy
from rclpy.node import Node
from rclpy.qos import QoSProfile, ReliabilityPolicy, HistoryPolicy
from std_msgs.msg import String


class Talker(Node):
    def __init__(self):
        super().__init__("talker")

        qos = QoSProfile(
            reliability=ReliabilityPolicy.RELIABLE,
            history=HistoryPolicy.KEEP_LAST,
            depth=10,
        )

        self.publisher_ = self.create_publisher(String, "chatter", qos)
        self.timer = self.create_timer(0.5, self.on_timer)
        self.count = 0

    def on_timer(self):
        msg = String()
        msg.data = f"hello ros2, count={self.count}"
        self.publisher_.publish(msg)
        self.get_logger().info(f"publish: {msg.data}")
        self.count += 1


def main():
    rclpy.init()
    node = Talker()
    try:
        rclpy.spin(node)
    finally:
        node.destroy_node()
        rclpy.shutdown()


if __name__ == "__main__":
    main()

2. Subscriber 節點

檔案位置：~/ros2_ws/src/my_package/my_package/listener.py

python
import rclpy
from rclpy.node import Node
from rclpy.qos import QoSProfile, ReliabilityPolicy, HistoryPolicy
from std_msgs.msg import String


class Listener(Node):
    def __init__(self):
        super().__init__("listener")

        qos = QoSProfile(
            reliability=ReliabilityPolicy.RELIABLE,
            history=HistoryPolicy.KEEP_LAST,
            depth=10,
        )

        self.subscription = self.create_subscription(
            String, "chatter", self.on_message, qos
        )

    def on_message(self, msg: String):
        self.get_logger().info(f"收到: {msg.data}")


def main():
    rclpy.init()
    node = Listener()
    try:
        rclpy.spin(node)
    finally:
        node.destroy_node()
        rclpy.shutdown()


if __name__ == "__main__":
    main()

記得在 setup.py 加上對應的 entry points，編譯後分別在兩個終端機執行：

bash
ros2 run my_package talker
ros2 run my_package listener

預期輸出

talker 終端機：

text
[INFO] [talker]: publish: hello ros2, count=0
[INFO] [talker]: publish: hello ros2, count=1
[INFO] [talker]: publish: hello ros2, count=2

listener 終端機：

text
[INFO] [listener]: 收到: hello ros2, count=0
[INFO] [listener]: 收到: hello ros2, count=1
[INFO] [listener]: 收到: hello ros2, count=2

用 CLI 直接觀察主題（不需要自己寫 subscriber 也能看）：

bash
ros2 topic echo /chatter

檢查兩端的 QoS 設定是否相容：

bash
ros2 topic info /chatter --verbose
text
Type: std_msgs/msg/String

Publisher count: 1

Node name: talker
QoS profile:
  Reliability: RELIABLE
  History (Depth): KEEP_LAST (10)

Subscription count: 1

Node name: listener
QoS profile:
  Reliability: RELIABLE
  History (Depth): KEEP_LAST (10)

建立自訂 message 型別

1. 建立獨立的介面套件

bash
cd ~/ros2_ws/src
ros2 pkg create --build-type ament_cmake my_package_msgs
mkdir my_package_msgs/msg

2. 定義 message 檔案

檔案位置：my_package_msgs/msg/Detection.msg

text
string label
float64 confidence
float64[4] bbox

3. 修改 CMakeLists.txt，加入介面產生規則

cmake
find_package(rosidl_default_generators REQUIRED)

rosidl_generate_interfaces(${PROJECT_NAME}
  "msg/Detection.msg"
)

4. 修改 package.xml

xml
<buildtool_depend>rosidl_default_generators</buildtool_depend>
<exec_depend>rosidl_default_runtime</exec_depend>
<member_of_group>rosidl_interface_packages</member_of_group>

5. 編譯並確認介面已產生

bash
colcon build --packages-select my_package_msgs
source install/setup.bash
ros2 interface show my_package_msgs/msg/Detection
text
string label
float64 confidence
float64[4] bbox

之後就能在其他套件的 Python 節點裡直接 from my_package_msgs.msg import Detection 使用。

常見錯誤與除錯技巧

錯誤一：Subscriber 完全收不到訊息，兩邊都沒有報錯

現象：ros2 topic list 能看到主題存在，talker 也持續在 log 裡顯示「已發布」，但 listener 就是沒有任何輸出，也沒有例外或錯誤訊息。

原因：幾乎可以肯定是 QoS 不相容——最常見的情境是 Publisher 用預設值（通常是 RELIABLE），但 Subscriber 明確要求 BEST_EFFORT 卻搭配了不相容的 Durability 設定，或反過來 Subscriber 要求 RELIABLE 而 Publisher 是 BEST_EFFORT。

排除方式：用 ros2 topic info /chatter --verbose 分別檢查發布端與訂閱端的 QoS profile，逐項比對 Reliability、Durability 是否相容。修正方式通常是讓 Subscriber 的要求「不要比 Publisher 提供的更嚴格」。

錯誤二：ModuleNotFoundError: No module named 'my_package_msgs'

text
ModuleNotFoundError: No module named 'my_package_msgs'

原因：自訂介面套件編譯完成後，忘記重新 source 工作空間；或是消費這個介面的節點所在套件，package.xml 裡沒有宣告對 my_package_msgs 的依賴，導致 rosdep／編譯順序沒有正確處理。

排除方式：

bash
source install/setup.bash
ros2 interface list | grep my_package_msgs

確認能列出來之後，檢查使用端套件的 package.xml 是否有加上：

xml
<depend>my_package_msgs</depend>

小結

主題通訊是多對多、鬆耦合的非同步模式，真正決定「訊息傳不傳得到」的關鍵在於 QoS 設定是否相容，而不是程式碼邏輯本身——這也是為什麼 QoS 不相容時完全沒有例外訊息，除錯時一定要養成用 ros2 topic info --verbose 檢查的習慣。下一節我們會進入「一問一答」型的同步通訊模式：服務（Service）。

延伸閱讀

常見問題

faq_01.log
Publisher 發布訊息時，如果沒有任何 Subscriber，訊息會怎麼樣？
訊息會直接被丟棄。ROS2 的主題通訊是「發布者不知道、也不關心誰在訂閱」的鬆耦合模式，沒有訂閱者代表沒有人接收，不會有佇列累積，也不會報錯。
faq_02.log
自訂 message 一定要放在獨立的套件裡嗎？
強烈建議如此。ROS2 的介面產生機制（rosidl）要求定義 message 的套件必須用 ament_cmake 建置，如果跟你的節點邏輯（可能是 ament_python）混在同一個套件會很麻煩，實務上幾乎都是另外建立一個像 my_package_msgs 的專用套件。
faq_03.log
QoS 設定不相容時，ROS2 會報錯嗎？
不會直接報錯，這是最容易讓人誤判的地方——Publisher 和 Subscriber 會各自建立成功，但底層 DDS 判定兩者不相容時，訊息就是靜默地無法傳遞。你必須主動用 ros2 topic info -v 或 ros2 doctor 檢查，才會發現問題。