如何在执行子进程回调期间维持ROS Publisher持续发布消息？

核心问题在于subprocess.run()是阻塞式调用，会占用回调函数的执行时间，导致ROS的事件循环无法及时处理其他任务，进而影响Publisher的正常工作。以下是两种可行的解决思路：

方法一：使用subprocess.Popen非阻塞启动子进程 #

subprocess.Popen启动子进程后会立即返回，不会阻塞当前回调函数，ROS主循环可以继续处理其他任务。如果不需要等待子进程完成就发布消息，直接调用即可；如果需要在子进程完成后再做后续处理，可以通过独立线程实现。

修改后的代码示例：

import subprocess
import rospy
import threading
from std_msgs.msg import String  # 补充缺失的消息类型导入

class Pub():
    def __init__(self):
        # 根据实际场景赋值以下路径参数
        self.planner_path = "path/to/your/planner.py"
        self.search_options = "your_search_alias"
        self.plan_path = "path/to/plan/file"
        self.domain_path = "path/to/domain.pddl"
        self.problem_path = "path/to/problem.pddl"

    def updateState(self, msg):
        cmd = ['python3', self.planner_path, "--alias", self.search_options, 
               "--plan-file", self.plan_path, self.domain_path, self.problem_path]

        # 非阻塞启动子进程
        proc = subprocess.Popen(cmd, shell=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        
        # 无需等待子进程完成，直接发布消息
        self.plan_pub.publish(msg)

        # 若需等待子进程完成后处理结果，启动独立线程（可选）
        threading.Thread(target=self._process_proc_result, args=(proc, msg)).start()

    def _process_proc_result(self, proc, msg):
        # 等待子进程结束并获取输出
        stdout, stderr = proc.communicate()
        rospy.loginfo(f"子进程执行完成，输出: {stdout.decode()}")
        # 可根据子进程输出修改msg后再发布
        # self.plan_pub.publish(modified_msg)

    def myPub(self):
        rospy.init_node('problem_formulator', anonymous=True)
        self.plan_pub = rospy.Publisher("plan", String, queue_size=10)
        # 修正回调函数引用，添加self
        rospy.Subscriber('model', String, self.updateState)
        rospy.spin()

if __name__ == "__main__":
    p_ = Pub()
    p_.myPub()

方法二：将子进程调用放入独立线程 #

直接把阻塞的subprocess.run()放到单独线程中执行，回调函数可以立即返回，ROS主循环不受阻塞影响。

修改后的代码示例：

import subprocess
import rospy
import threading
from std_msgs.msg import String

class Pub():
    def __init__(self):
        self.planner_path = "path/to/your/planner.py"
        self.search_options = "your_search_alias"
        self.plan_path = "path/to/plan/file"
        self.domain_path = "path/to/domain.pddl"
        self.problem_path = "path/to/problem.pddl"

    def _execute_planner(self, msg):
        cmd = ['python3', self.planner_path, "--alias", self.search_options, 
               "--plan-file", self.plan_path, self.domain_path, self.problem_path]
        # 阻塞执行子进程，因在独立线程中，不影响ROS主循环
        result = subprocess.run(cmd, shell=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        rospy.loginfo(f"规划器执行完成，返回码: {result.returncode}")
        # 子进程完成后发布消息
        self.plan_pub.publish(msg)

    def updateState(self, msg):
        # 启动守护线程执行子进程调用
        threading.Thread(target=self._execute_planner, args=(msg,), daemon=True).start()

    def myPub(self):
        rospy.init_node('problem_formulator', anonymous=True)
        self.plan_pub = rospy.Publisher("plan", String, queue_size=10)
        rospy.Subscriber('model', String, self.updateState)
        rospy.spin()

if __name__ == "__main__":
    p_ = Pub()
    p_.myPub()

额外注意事项 #

• 原代码中rospy.Subscriber的回调函数缺少self，会导致无法找到类方法，必须修正为self.updateState。
• 必须导入from std_msgs.msg import String，否则无法识别消息类型。
• 如果子进程会产生大量输出，建议及时处理stdout和stderr，避免缓冲区溢出。

内容的提问来源于stack exchange，提问作者Bilal