第4章:理解ROS服务与动作:Service与Action机制

说到ROS的通信机制,很多人第一反应就是Topic。没错,Topic确实是最常用的。但我在实际项目中遇到过不少场景——光靠Topic根本搞不定。比如你让机器人抓个杯子,你发个指令过去,然后呢?它抓没抓到?抓了多久?中间能不能取消?这些Topic都回答不了。

这时候就需要Service和Action出场了。说白了,它们就是ROS里两种「有反馈」的通信方式。今天我就带你把这俩东西彻底搞明白。

4.1 Service:一问一答的同步通信

Service的机制很简单——客户端发一个请求,服务端处理完,返回一个响应。就像你问路,别人告诉你答案,然后对话结束。

我习惯把Service比作「函数调用」。你传参数进去,拿到返回值。只不过这个函数跑在另一个节点里。

4.1.1 什么时候用Service?

  • 需要立即得到结果的任务
  • 任务执行时间很短(毫秒级)
  • 不需要中途取消或监控进度

举个例子:你让机器人打开一盏灯。发个请求,灯亮了,返回成功。就这么简单。

注意:Service是阻塞的。客户端发完请求后,会一直等响应。如果服务端处理时间太长,整个节点都可能卡住。我曾经在一个项目里用Service控制电机,结果电机响应慢了3秒,整个控制循环都崩了。后来全改成了Action。

4.1.2 自定义srv文件

Service的数据类型由.srv文件定义。格式很简单:请求部分和响应部分用---隔开。

# SetMotorSpeed.srv
# 请求部分
string motor_name
float64 target_speed

---
# 响应部分
bool success
string message

我个人习惯把srv文件放在功能包里的srv/目录下。然后在CMakeLists.txt里加上:

add_service_files(
  FILES
  SetMotorSpeed.srv
)

嗯,这里要注意——别忘了在package.xml里声明对message_generationmessage_runtime的依赖。

4.1.3 Service Server实战

写一个Service Server其实就三步:创建服务、处理请求、返回响应。

# motor_service_server.py
import rospy
from your_package.srv import SetMotorSpeed, SetMotorSpeedResponse

def handle_motor_speed(req):
    rospy.loginfo(f"收到请求:设置 {req.motor_name} 速度为 {req.target_speed}")
    
    # 这里假装在控制电机
    if req.target_speed < 0:
        return SetMotorSpeedResponse(False, "速度不能为负数")
    
    # 实际项目中,这里会调用硬件驱动
    success = True
    message = f"{req.motor_name} 速度已设置为 {req.target_speed}"
    
    return SetMotorSpeedResponse(success, message)

def motor_server():
    rospy.init_node('motor_service_server')
    service = rospy.Service('set_motor_speed', SetMotorSpeed, handle_motor_speed)
    rospy.loginfo("电机速度控制服务已启动")
    rospy.spin()

if __name__ == '__main__':
    motor_server()

4.1.4 Service Client实战

客户端更简单——等服务就绪,然后调用。

# motor_service_client.py
import rospy
from your_package.srv import SetMotorSpeed

def call_motor_service():
    rospy.init_node('motor_service_client')
    
    # 等待服务就绪
    rospy.wait_for_service('set_motor_speed')
    
    try:
        # 创建服务代理
        set_speed = rospy.ServiceProxy('set_motor_speed', SetMotorSpeed)
        
        # 调用服务
        response = set_speed('left_wheel', 1.5)
        
        if response.success:
            rospy.loginfo(f"成功:{response.message}")
        else:
            rospy.logerr(f"失败:{response.message}")
            
    except rospy.ServiceException as e:
        rospy.logerr(f"服务调用失败:{e}")

if __name__ == '__main__':
    call_motor_service()

小技巧:我习惯在客户端里加一个超时机制。用rospy.wait_for_service('service_name', timeout=5),避免无限等待。这在机器人启动顺序不确定时特别有用。

4.2 Action:带反馈的异步通信

Action比Service复杂,但也更强大。它支持三个核心功能:

  • 发送目标(Goal)
  • 接收实时反馈(Feedback)
  • 中途取消或查询状态

你想想看,让机器人导航到某个点——这个过程可能要几十秒。你总不能让客户端一直傻等着吧?Action就是为这种场景设计的。

4.2.1 自定义action文件

Action定义分三部分:目标、结果、反馈。用---分隔。

# NavigateToPose.action
# 目标(Goal)
geometry_msgs/Pose2D target_pose
float64 timeout

---
# 结果(Result)
bool reached
float64 distance_traveled
float64 time_elapsed

---
# 反馈(Feedback)
float64 distance_remaining
float64 progress_percent
string status_description

文件放在功能包的action/目录下。CMakeLists.txt里要加:

add_action_files(
  FILES
  NavigateToPose.action
)

4.2.2 Action Server实战

写Action Server比Service复杂不少。我直接给你一个完整的例子:

# navigation_action_server.py
import rospy
import actionlib
from your_package.msg import NavigateToPoseAction, NavigateToPoseFeedback, NavigateToPoseResult

class NavigationServer:
    def __init__(self):
        self.server = actionlib.SimpleActionServer(
            'navigate_to_pose', 
            NavigateToPoseAction, 
            self.execute_callback, 
            False
        )
        self.server.start()
        rospy.loginfo("导航动作服务器已启动")
    
    def execute_callback(self, goal):
        rospy.loginfo(f"收到导航目标:x={goal.target_pose.x}, y={goal.target_pose.y}")
        
        # 创建反馈和结果对象
        feedback = NavigateToPoseFeedback()
        result = NavigateToPoseResult()
        
        # 模拟导航过程
        total_distance = 10.0  # 假设总距离10米
        traveled = 0.0
        
        rate = rospy.Rate(10)  # 10Hz反馈
        
        while traveled < total_distance:
            # 检查是否被取消
            if self.server.is_preempt_requested():
                rospy.loginfo("导航被取消")
                self.server.set_preempted()
                return
            
            # 模拟移动
            traveled += 0.5
            remaining = total_distance - traveled
            
            # 发布反馈
            feedback.distance_remaining = remaining
            feedback.progress_percent = (traveled / total_distance) * 100.0
            feedback.status_description = f"已前进 {traveled:.1f} 米"
            self.server.publish_feedback(feedback)
            
            rate.sleep()
        
        # 设置结果
        result.reached = True
        result.distance_traveled = traveled
        result.time_elapsed = traveled / 0.5  # 粗略计算
        
        self.server.set_succeeded(result)

if __name__ == '__main__':
    rospy.init_node('navigation_action_server')
    server = NavigationServer()
    rospy.spin()

关键点:Action Server里一定要处理is_preempt_requested()。我曾经没加这个检查,结果客户端取消请求后,机器人还在傻乎乎地往前跑,差点撞墙。血的教训啊。

4.2.3 Action Client实战

客户端这边,我习惯用SimpleActionClient,它封装了很多细节。

# navigation_action_client.py
import rospy
import actionlib
from your_package.msg import NavigateToPoseAction, NavigateToPoseGoal

def feedback_callback(feedback):
    rospy.loginfo(f"进度:{feedback.progress_percent:.1f}% - {feedback.status_description}")

def send_navigation_goal():
    rospy.init_node('navigation_action_client')
    
    # 创建客户端
    client = actionlib.SimpleActionClient('navigate_to_pose', NavigateToPoseAction)
    
    # 等待服务器
    rospy.loginfo("等待导航服务器...")
    client.wait_for_server()
    rospy.loginfo("导航服务器已连接")
    
    # 创建目标
    goal = NavigateToPoseGoal()
    goal.target_pose.x = 5.0
    goal.target_pose.y = 3.0
    goal.timeout = 30.0
    
    # 发送目标,注册反馈回调
    client.send_goal(goal, feedback_cb=feedback_callback)
    
    # 等待结果(带超时)
    finished = client.wait_for_result(rospy.Duration.from_sec(60.0))
    
    if finished:
        result = client.get_result()
        if result.reached:
            rospy.loginfo(f"导航成功!行驶距离:{result.distance_traveled:.2f}米")
        else:
            rospy.logwarn("导航未到达目标")
    else:
        rospy.logerr("导航超时")
        client.cancel_goal()

if __name__ == '__main__':
    send_navigation_goal()

4.3 Service vs Action:怎么选?

很多新手会纠结这个问题。我直接给你一个对照表:

特性 Service Action
通信模式 同步(一问一答) 异步(可监控进度)
执行时间 短(毫秒级) 长(秒级甚至更久)
中途取消 不支持 支持
实时反馈 有(Feedback)
适用场景 开关控制、参数查询 导航、机械臂运动、复杂任务

我的经验法则很简单:如果任务执行时间超过1秒,或者你需要知道中间状态,就用Action。否则用Service就够了。

对了,还有个小细节——Action底层其实也是用Topic实现的。它内部用了三个Topic:一个发目标,一个发反馈,一个发结果。但你别管这些,直接用actionlib库就好。

4.4 实战中的坑与技巧

最后分享几个我踩过的坑:

  1. Service超时问题:默认情况下,Service调用没有超时。我建议客户端里手动加一个超时检测,用rospy.Time.now()计时。
  2. Action的预emption:一定要在Server端正确处理is_preempt_requested()。否则客户端取消不了,用户体验极差。
  3. 反馈频率:Action的Feedback不要发太频繁。10Hz就够了,发太快会占用大量带宽。我见过有人用100Hz发反馈,结果CPU飙到80%。
  4. 自定义消息的路径:srv和action文件里引用的其他消息类型,一定要写完整的包名。比如geometry_msgs/Pose2D,别漏了。

好了,Service和Action的核心内容就这些。说白了,Service适合「快问快答」,Action适合「长期任务」。你只要记住这个原则,大部分场景都不会选错。

下一章我们聊聊ROS的Parameter Server和动态参数配置。到时候我会分享一个我在调试激光雷达时用参数动态调整的实战案例,挺有意思的。