【ROS2入門④】トピック通信実装 - Pythonでパブリッシャーとサブスクライバーをつくる

ROS 2 基本概念を理解した状態から、Python で Publisher / Subscriber を書き、colcon でビルドして ros2 run まで動かします。公式 Writing a simple publisher and subscriber (Python) をベースに、setup.py の console_scripts など詰まりやすい設定を補足します。2026年5月時点、Ubuntu 22.04 + ROS 2 Humble を前提としています。

Publisher/Subscriber 実装の事前準備：ワークスペース作成

ワークスペースを作成します。

mkdir -p ~/ros2_ws/src
cd ~/ros2_ws/src
ros2 pkg create --build-type ament_python my_topic_pkg

ros2 pkg create の --node-name は自動でノードひな型を 1つ作ります。Publisher と Subscriber を両方置きたい場合は、my_topic_pkg/ 配下へ publisher_node.py と subscriber_node.py を新規ファイルとして追加します（名前はワークスペースに合わせてよいです）。

作成されるディレクトリ構成（例）：

my_topic_pkg/
├── my_topic_pkg/
│   ├── __init__.py
│   ├── publisher_node.py
│   └── subscriber_node.py
├── package.xml
├── setup.cfg
└── setup.py

Python で Publisher（送信側）を実装する方法

# publisher_node.py
import rclpy
from rclpy.node import Node
from std_msgs.msg import String


class MinimalPublisher(Node):

    def __init__(self):
        super().__init__('minimal_publisher')
        # Publisher作成: メッセージ型, トピック名, QoSキューサイズ
        self.publisher_ = self.create_publisher(String, 'my_topic', 10)
        # 0.5秒ごとにコールバックを呼び出すタイマー
        self.timer = self.create_timer(0.5, self.timer_callback)
        self.i = 0

    def timer_callback(self):
        msg = String()
        msg.data = f'Hello ROS2: {self.i}'
        self.publisher_.publish(msg)
        self.get_logger().info(f'Publishing: "{msg.data}"')
        self.i += 1


def main(args=None):
    rclpy.init(args=args)
    node = MinimalPublisher()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()

Publisherの仕組み

Python で Subscriber（受信側）を実装する方法

# subscriber_node.py
import rclpy
from rclpy.node import Node
from std_msgs.msg import String


class MinimalSubscriber(Node):

    def __init__(self):
        super().__init__('minimal_subscriber')
        self.subscription = self.create_subscription(
            String,
            'my_topic',
            self.listener_callback,
            10)

    def listener_callback(self, msg):
        self.get_logger().info(f'Received: "{msg.data}"')


def main(args=None):
    rclpy.init(args=args)
    node = MinimalSubscriber()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()

setup.py でエントリーポイントを登録する方法

ament_python ではパッケージ名を変えたら、package.xml の name（XMLの name フィールド）、setup.py の package_name と packages=[...]、setup.cfg があればそこにある packages 指定、console_scripts のモジュールパスをまとめて追います。どれか一つだけ直すと ros2 run が「見えるが動かない」状態になりやすいです。

# setup.py
from setuptools import setup

package_name = 'my_topic_pkg'

setup(
    name=package_name,
    version='0.0.1',
    packages=[package_name],
    install_requires=['setuptools'],
    entry_points={
        'console_scripts': [
            'publisher = my_topic_pkg.publisher_node:main',
            'subscriber = my_topic_pkg.subscriber_node:main',
        ],
    },
)

colcon ビルドと Publisher/Subscriber の実行手順

cd ~/ros2_ws
colcon build --packages-select my_topic_pkg
. install/setup.bash

ターミナル1 - Publisher起動:

ros2 run my_topic_pkg publisher
# [INFO] [minimal_publisher]: Publishing: "Hello ROS2: 0"
# [INFO] [minimal_publisher]: Publishing: "Hello ROS2: 1"

ターミナル2 - Subscriber起動:

ros2 run my_topic_pkg subscriber
# [INFO] [minimal_subscriber]: Received: "Hello ROS2: 5"
# [INFO] [minimal_subscriber]: Received: "Hello ROS2: 6"

トピック通信を確認する CLI コマンド

# トピック一覧
ros2 topic list
# /my_topic

# トピックのデータをリアルタイム表示
ros2 topic echo /my_topic

# 送信頻度確認
ros2 topic hz /my_topic
# average rate: 2.000

QoS（Quality of Service）とは？入門レベルの設定

QoS（Quality of Service）は通信の約束ごと。デフォルトでも動くことが多いですが、画像や LiDAR では BEST_EFFORT などを検討する場面があります。About QoS を参照。

信頼性 (Reliability):
  RELIABLE     … なるべく届ける前提（再送回数などにも依存）
  BEST_EFFORT … 取りこぼし許容・レイテンシ優先になりやすい

デュラビリティ Durability:
  VOLATILE         … 購読側がいないときの過去ログは狙わないことが多い
  TRANSIENT_LOCAL … 「最新」を保持したいモデルがあります（用途次第）

from rclpy.qos import QoSProfile, ReliabilityPolicy

qos = QoSProfile(reliability=ReliabilityPolicy.RELIABLE, depth=10)
self.publisher_ = self.create_publisher(String, 'my_topic', qos)

トピック通信実装のまとめ

create_publisher / create_subscription、spin とタイマーの組み合わせが基本形
setup.py の console_scripts で ros2 run できる名前を定義する
QoS と API の細部はディストリと rclpy の版で差が出る。エラーになったらメッセージ型と QoS プロファイルを先に確認
シリーズ続きではサービス／アクションを Python で触る記事へ進む想定です。

検証環境とつまずきやすい点

デスクトップ上の colcon build と 2 ターミナル実行での確認メモです。TB3 実機とは無関係なので、トピック名やノード名は自分のワークスペースに合わせてください。

ROS 2 が環境読み込み（source install/setup.bash）されていないと ros2 コマンドやパッケージが見つからない
パッケージ名変更時は setup.cfg / setup.py / package.xml の対応関係がズレやすい

30代エンジニアの備忘録