夜雨聆风目标： 使用 AsyncNode（基于 asyncio 的原生节点 API）创建并运行一个服务和客户端。
教程级别： 中级
时间： 20 分钟
背景
AsyncNode 是初学者教程中使用的基于执行器的 Node 的 asyncio 原生替代方案。这允许您在 ROS 2 回调内部 await 其他异步操作，而不会阻塞节点的其余部分。
除了服务和客户端之外，AsyncNode 还支持定时器、订阅和发布者。回调可以是 async def 协程或常规 def 函数。请记住，同步回调不能 await 任何东西。
前置条件
- 您应该已完成初学者服务和客户端教程。
- 您应该熟悉基本的 asyncio 概念，如 async def 和 await 以及 asyncio.run()。
- AsyncNode 目前位于 rclpy.experimental 中，需要 Python 3.12 或更新版本。
任务
1 创建包
打开一个新终端并加载您的 ROS 2 安装，以便 ros2 命令能够工作。
导航到在之前教程中创建的 ros2_ws/src 目录，并创建一个新包：
ros2 pkg create --build-type ament_python python_async_node --dependencies rclpy example_interfaces2 编写服务节点
在 ros2_ws/src/python_async_node/python_async_node 目录中，创建一个名为 async_service.py 的新文件，并粘贴以下代码：
import asyncio
from example_interfaces.srv import Trigger
import rclpy
from rclpy.experimental import AsyncNode
class TriggerServer(AsyncNode):
def __init__(self):
super().__init__('trigger_server')
self._service = self.create_service(
Trigger, 'trigger', self._callback
)
async def _callback(self, _request, response):
self.get_logger().info('Incoming trigger request')
await self.get_clock().sleep(2.0)
response.success = True
response.message = 'Slept for 2.0s'
return response
async def _async_main():
with rclpy.init():
node = TriggerServer()
await node.run()
def main():
asyncio.run(_async_main())
if __name__ == '__main__':
main()2.1 检查代码
第一个 import 语句拉入了 asyncio、rclpy、从 rclpy.experimental 导入的 AsyncNode 类，以及 Trigger 服务类型。
import asyncio
from example_interfaces.srv import Trigger
import rclpy
from rclpy.experimental import AsyncNodeTriggerServer 类继承自 AsyncNode 而不是 Node。其构造函数将节点初始化为名称 trigger_server 并创建一个服务。
class TriggerServer(AsyncNode):
def __init__(self):
super().__init__('trigger_server')
self._service = self.create_service(
Trigger, 'trigger', self._callback
)服务回调是一个协程。在其中您可以 await 任何异步操作——这里节点调用 self.get_clock().sleep(2.0)，类似于 asyncio.sleep，但支持 ROS 仿真时间。当此协程被挂起时，节点仍然对其他回调保持响应。
async def _callback(self, _request, response):
self.get_logger().info('Incoming trigger request')
await self.get_clock().sleep(2.0)
response.success = True
response.message = 'Slept for 2.0s'
return response文件底部定义了一个 async 辅助函数，用于初始化 rclpy、构造节点并运行它：
async def _async_main():
with rclpy.init():
node = TriggerServer()
await node.run()要运行 AsyncNode，您需要 await node.run()——相当于 rclpy.spin(node)。
最后，main 本身是一个常规同步函数，以便 ros2 run 可以将其用作入口点——ros2 run 将入口点作为常规函数调用，因此它不能是 async def。它只是调用 asyncio.run 来执行 _async_main 协程：
def main():
asyncio.run(_async_main())3 编写客户端节点
在 ros2_ws/src/python_async_node/python_async_node 目录中，创建一个名为 async_client.py 的新文件，并粘贴以下代码：
import asyncio
from example_interfaces.srv import Trigger
import rclpy
from rclpy.experimental import AsyncNode
async def _async_main():
with rclpy.init():
async with AsyncNode('trigger_client') as node:
client = node.create_client(Trigger, 'trigger')
node.get_logger().info('Waiting for trigger service...')
await client.wait_for_service()
node.get_logger().info('Service is available, sending request')
request = Trigger.Request()
response = await client.call(request)
node.get_logger().info(
f'Service responded: success={response.success}, '
f'message="{response.message}"'
)
def main():
asyncio.run(_async_main())
if __name__ == '__main__':
main()3.1 检查代码
与服务代码一样，import 语句拉入了 asyncio、rclpy、AsyncNode 和 Trigger 类型。
客户端使用了 AsyncNode 的第二个入口点：async with AsyncNode(...) as node:。进入 async with 块实际上就是运行节点——类似于服务端的 await node.run()。此时，节点上已创建的任何订阅、服务和定时器都开始分派回调。节点在块的持续时间内保持运行，当块退出时，自动调用 destroy_node() 并销毁所有实体。这对于像一次性客户端这样的短程序来说是合适的选择。
async with AsyncNode('trigger_client') as node:
client = node.create_client(Trigger, 'trigger')在发送请求之前，await client.wait_for_service() 挂起协程，直到发现匹配的服务（如果已经存在则立即返回）——而不阻塞事件循环，因此节点上的其他回调继续运行。这避免了“客户端在服务器之前启动”的失败模式，即请求否则会被发送到虚空之中。
node.get_logger().info('Waiting for trigger service...')
await client.wait_for_service()
node.get_logger().info('Service is available, sending request')实际的服务调用只有一行：
request = Trigger.Request()
response = await client.call(request)使用 AsyncNode，响应直接被 await——没有 Future，并且在调用进行时节点继续处理其他工作。
4 添加入口点
为了允许 ros2 run 命令运行您的节点，请将入口点添加到位于 ros2_ws/src/python_async_node 目录中的 setup.py。
setup.py 的 entry_points 字段应如下所示：
entry_points={
'console_scripts': [
'service = python_async_node.async_service:main',
'client = python_async_node.async_client:main',
],
},5 构建并运行
最佳实践是在工作空间根目录（ros2_ws）运行 rosdep 来检查构建前是否缺少依赖项：
rosdep install -i --from-path src --rosdistro lyrical -y导航回工作空间根目录 ros2_ws，并构建您的新包：
colcon build --symlink-install --packages-select python_async_node--symlink-install 标志将 Python 源文件符号链接到安装空间而不是复制它们，因此稍后的编辑无需重新构建即可生效。
打开一个新终端，导航到 ros2_ws，并加载设置文件：
source install/setup.bash现在运行服务节点：
ros2 run python_async_node service该节点将等待客户端的请求。
打开另一个终端，再次从 ros2_ws 内部加载设置文件。启动客户端节点：
ros2 run python_async_node client
[INFO] [trigger_client]: Service responded: success=True, message="Slept for 2.0s"返回到服务节点正在运行的终端。您将看到它记录了传入的请求：
[INFO] [trigger_server]: Incoming trigger request在服务端终端中按 Ctrl-C 停止节点。
扩展
上面的基本 Trigger 服务器使用 await clock.sleep(2.0) 作为异步工作的示例。下面展示两个更实际的使用场景——两者都是对 async_service.py 的直接修改。
由于包是用 --symlink-install 构建的，对 Python 文件的编辑无需重新构建即可生效——用 Ctrl-C 停止正在运行的服务，然后重新启动以应用更改。
从回调中执行阻塞函数
loop.call_soon_threadsafe 或 asyncio.run_coroutine_threadsafe 将该工作调度回事件循环线程。许多有用的库仍然使用同步 API 编写。直接从异步回调中调用它们会阻塞事件循环，并使节点上的所有其他回调停滞。为了解决这个问题，asyncio.to_thread 在工作线程上运行同步函数并返回一个可等待对象，使节点保持响应。
例如，我们可以通过 asyncio.to_thread 从 GitHub 发起同步 HTTP 请求以获取最新的 ROS 2 发布标签。用以下内容替换 async_service.py 的主体：
import asyncio
import json
import urllib.request
from example_interfaces.srv import Trigger
import rclpy
from rclpy.experimental import AsyncNode
ROS2_LATEST_RELEASE_URL = 'https://api.github.com/repos/ros2/ros2/releases/latest'
class TriggerServer(AsyncNode):
def __init__(self):
super().__init__('trigger_server')
self._service = self.create_service(
Trigger, 'trigger', self._callback
)
def _fetch_latest_release(self):
self.get_logger().info('Fetching latest ROS 2 release from GitHub')
with urllib.request.urlopen(ROS2_LATEST_RELEASE_URL) as resp:
return json.loads(resp.read())['tag_name']
async def _callback(self, _request, response):
tag = await asyncio.to_thread(self._fetch_latest_release)
response.success = True
response.message = f'Latest ROS 2 release: {tag}'
return response
async def _async_main():
with rclpy.init():
node = TriggerServer()
await node.run()
def main():
asyncio.run(_async_main())
if __name__ == '__main__':
main()在一个终端中重新启动服务，然后在另一个终端中运行客户端：
ros2 run python_async_node service
ros2 run python_async_node client
[INFO] [trigger_client]: Service responded: success=True, message="Latest ROS 2 release: <release-tag>"从回调内部调用另一个服务
由于回调是协程，它们可以像任何其他异步操作一样 await 其他 ROS 2 服务调用。
替换 async_service.py 的主体，使 Trigger 回调委托给 demo_nodes_py 的 add_two_ints 服务：
import asyncio
from example_interfaces.srv import AddTwoInts, Trigger
import rclpy
from rclpy.experimental import AsyncNode
class TriggerServer(AsyncNode):
def __init__(self):
super().__init__('trigger_server')
self._client = self.create_client(AddTwoInts, 'add_two_ints')
self._service = self.create_service(
Trigger, 'trigger', self._callback
)
async def _callback(self, _request, response):
self.get_logger().info('Waiting for add_two_ints service...')
await self._client.wait_for_service()
self.get_logger().info('Calling add_two_ints with a=2, b=2')
nested_request = AddTwoInts.Request(a=2, b=2)
nested_response = await self._client.call(nested_request)
response.success = True
response.message = f'2+2={nested_response.sum}'
return response
async def _async_main():
with rclpy.init():
node = TriggerServer()
await node.run()
def main():
asyncio.run(_async_main())
if __name__ == '__main__':
main()客户端在 __init__ 中创建并在回调内部使用。由于调用被 await，协程不会阻塞节点——在等待 add_two_ints 响应时，其他回调继续取得进展。
await client.call(request) 可以在任何回调中工作，无需额外配置。在一个终端中运行底层的 add_two_ints 服务器：
ros2 run demo_nodes_py add_two_ints_server重新启动您的服务并像之前一样运行客户端：
ros2 run python_async_node service
ros2 run python_async_node client
[INFO] [trigger_client]: Service responded: success=True, message="2+2=4"并发处理请求
默认情况下，异步服务一次处理一个请求——如果回调仍在运行时新请求到达，新请求将等待。
传递 concurrent=True 会在其自己的任务上分派每个传入请求，使多个回调能够并行运行：
self._service = self.create_service(
Trigger, 'trigger', self._callback, concurrent=True)使用并发分派，同时到达的三个客户端每个都在大约两秒后获得响应，而不是排队等待。同样的标志也可用于 create_subscription。
总结
您使用 AsyncNode 编写了一个异步服务和客户端，使用可以 await 长时间运行操作的 async def 回调。由于 AsyncNode 在 asyncio 上运行，您的节点可以直接使用 Python 异步生态系统的其余部分——Web 框架、消息传递客户端、数据库驱动等。
