> ## Documentation Index
> Fetch the complete documentation index at: https://docs.wuji.tech/llms.txt
> Use this file to discover all available pages before exploring further.

# 数据订阅：异步

## 概述

用 `async/await` 订阅 Wuji Hand 2 的 `joint_states`。与回调式相比，数据在协程里按顺序取出，不需要跨线程传递。

应用本身已经跑在 asyncio 上时，这种方式更自然。多设备并发用 `asyncio.gather()` 就能表达，不用自己管线程。

对应示例：`examples/python/wuji_hand_2/1.subscribe_async.py`。

关节状态与各字段定义见 [SDK 接口](https://docs.wuji.tech/docs/zh/wuji-hand/beta1/sdk-reference.md)，本页只讲怎么跑起来与实测结果。

> [!NOTE]
>
> C 接口没有对应的异步形式。C 侧订阅统一走回调，见[数据订阅：回调](https://docs.wuji.tech/docs/zh/wuji-hand/beta1/sdk-practice/subscribe-callback)。

本示例只读取，不使能电机，手指不会动。

## 前置条件

### 硬件

| 项目 | 要求 |
| - | - |
| Wuji Hand 2 | 1 只 |
| 连接方式 | 以太网 |

### 软件

| 项目 | 要求 |
| - | - |
| 操作系统 | Ubuntu 22.04 LTS 及以上版本 |
| Wuji SDK | v2026.8.31 或更高版本 |
| Python | 3.10 及以上版本 |

```bash theme={null}
pip install wuji-sdk
```

## 数据流

```mermaid theme={null}
flowchart TD
    FW["Wuji Hand 2 固件<br/>约 1 kHz 广播关节状态"]
    SUB["joint_states().subscribe()<br/>返回订阅句柄，内部维护队列"]
    LOOP["asyncio 事件循环"]
    AWAIT["await sub.recv_async()<br/>队列为空时挂起协程"]
    APP["应用协程<br/>按帧处理"]

    FW -->|"帧入队"| SUB
    SUB --> LOOP
    LOOP --> AWAIT
    AWAIT -->|"JointStateFrame"| APP
    APP -->|"继续等下一帧"| AWAIT
```

`subscribe()` 拿到的句柄同时支持两种取法：`await recv_async()` 异步取，`recv()` 同步取。帧结构与回调式完全一致。

## 运行时分支

| 分支点 | 表现 | 处理 |
| - | - | - |
| 队列为空 | `recv_async()` 挂起当前协程，让出事件循环 | 正常行为，不要用忙等 |
| 同步 `recv()` 队列为空 | 立即返回 `None`，不阻塞 | 循环里配合 `recv()` 要自己控制节奏 |
| 设备断连 | `recv_async()` 抛 `WujiException` | 立即抛出异常 |
| 多台设备 | 示例连接扫到的每一台，用 `asyncio.gather()` 并发 | 共用网络要改成指定 SN |
| 协程被取消 | `gather` 抛出，进入 `finally` 断开设备 | 示例靠 `finally` 里的 `disconnect_all()` 收尾 |

## 调用顺序

```python theme={null}
import asyncio
from wuji_sdk import SdkManager, DeviceType, JointStateFrame

async def print_joint_state(hand):
    # 1. 取订阅句柄
    sub = hand.joint_states().subscribe()
    while True:
        # 2. 异步等下一帧，队列空时协程挂起
        frame: JointStateFrame = await sub.recv_async()
        h = frame.header
        positions = [f"{j.position:+.3f}" for j in frame.joints[:4]]
        print(f"[{hand.device_name}][JointState] seq={h.seq} "
              f"t={h.timestamp_us}us frame={h.frame_id!r} thumb_pos={positions}")

async def main():
    manager = SdkManager.instance()
    hands = [d for d in manager.scan() if d.device_type == DeviceType.WujiHand2]
    try:
        # 3. 连接。多设备环境指定 sn
        hand = manager.connect(sn=hands[0].sn, device_name="hand_0")
        # 4. 每台设备一个协程，并发跑
        await asyncio.gather(print_joint_state(hand))
    finally:
        manager.disconnect_all()

asyncio.run(main())
```

订阅是纯读取，不需要 `enable()` 与 `disable()`。

## 运行与输出

```bash theme={null}
python examples/python/wuji_hand_2/1.subscribe_async.py
```

```text theme={null}
Connected: WH2KA00XXXXXXXXX (20 joints online)
[hand_0][JointState] seq=147508 t=210677379743us frame='r_wrist' thumb_pos=['+0.003', '-0.018', '+0.021', '+0.056']
```

### 实测数据

在 20 关节全在线的右手 Wuji Hand 2 上实跑：

| 指标 | 实测 |
| - | - |
| `recv_async()` 吞吐 | **1088 Hz**（3 秒 3265 帧） |
| 同一句柄 `recv()` 同步取 | 1 秒 1091 帧 |
| `frame_id` | `r_wrist` |
| `joints` 长度 | 20 |

异步与同步两种取法的吞吐一致，都跟得上设备约 1 kHz 的广播。选哪种取决于应用结构，不是性能问题。

## 调整与适配

### 与回调式的取舍

| | 异步 `recv_async()` | 回调 `subscribe_with_callback()` |
| - | - | - |
| 执行位置 | 调用方的事件循环 | SDK 工作线程 |
| 背压 | 队列积压可感知，可主动丢帧 | 回调阻塞会直接拖慢工作线程 |
| 适用 | 应用已在 asyncio 上 | 简单转发、写队列 |
| C 接口 | 无 | 有 |

### 多设备并发

每台设备一个协程，`asyncio.gather()` 并发即可。设备之间互不阻塞。

### 降低输出率

约 1 kHz 对多数应用偏高。用同一句柄的 `set_rate()` 降频，注意有固件版本要求，见[输出率调节](https://docs.wuji.tech/docs/zh/wuji-hand/beta1/sdk-practice/stream-rate)。

## 相关内容

* [数据订阅：回调](https://docs.wuji.tech/docs/zh/wuji-hand/beta1/sdk-practice/subscribe-callback)：回调式写法，以及 C 接口的唯一形式
* [输出率调节](https://docs.wuji.tech/docs/zh/wuji-hand/beta1/sdk-practice/stream-rate)：降低固件推送频率
* [数据订阅机制](https://docs.wuji.tech/docs/zh/wuji-sdk/latest/data-subscription.md)：订阅模型与生命周期


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.