企业官网建设流程全解析

Ray 新手性能四戒：延迟 ray.get、避免微任务、复用对象引用与流水线处理

【免费下载链接】rayRay is an AI compute engine. Ray consists of a core distributed runtime and a set of AI Libraries for accelerating ML workloads.项目地址: https://gitcode.com/gh_mirrors/ra/ray

本篇文章围绕 Ray 官方文档《Tips for first-time users》（即仓库中的 tips-for-first-time.rst）展开，面向第一次接触 Ray 的开发者，系统讲解四个最容易犯、却会显著拖慢程序性能的常见错误：过早调用ray.get()、切分过小的远程任务、重复向远程任务传同一大对象、以及"等全部结果就绪再处理"的同步式数据处理方式。读完本文，你将掌握ray.init()、@ray.remote、.remote()、ray.put()、ray.get()、ray.wait()六个核心 API 的正确使用节奏，并能用一套可量化的方法评估自己的任务粒度是否合适。

Ray 提供了高度灵活、却又极简易用的 API，但恰恰是这种"把函数加个装饰器就能并行"的低门槛，容易让新手写出"看似并行、实则串行"的代码。文中的四个 Tip 正是针对这些问题给出的实战纪律，每个 Tip 都配有可直接运行的完整示例、真实耗时数据以及源码层面的解释。

核心 API 速查表

本文所有示例都只依赖以下六个核心 API，先建立整体认知：

从源码看，ray.get、ray.put、ray.wait三个核心函数的 Python 层实现集中在 python/ray/_private/worker.py 中（get位于 L2854–L3032，put位于 L3037–L3084，wait位于 L3093 之后）。其中ray.get的文档字符串明确标注：该方法会一直阻塞到对象在本地对象存储中可用为止；若传入的是对象 ID 列表，则保持输入顺序返回对应对象列表。这一点对理解 Tip 1 和 Tip 4 至关重要。

文中所有性能数据均来自原文档在 13 英寸 MacBook Pro（2.7 GHz Core i7、16GB 内存）上的实测结果。为避免机器差异带来的波动，示例统一使用ray.init(num_cpus=4)指定 4 个 CPU：由于每个任务默认申请 1 个 CPU，这个设置允许最多 4 个任务并行执行，系统构成为 1 个 driver（驱动程序）加最多 4 个 worker。

Tip 1：延迟调用 ray.get()，把阻塞留在最后

Ray 中所有远程操作（任务、Actor 方法）的调用都是异步的：调用会立即返回一个 future（本质上是一个结果 ID / ObjectRef），这正是并行得以实现的关键——driver 可以在不等待结果的情况下批量发起多个操作。要拿到真正的结果，必须对结果 ID 调用ray.get()，而该调用会阻塞直到结果可用。副作用是：阻塞期间 driver 也无法发起新的远程调用，从而拖累并行度。

遗憾的是，新手"顺手"就写ray.get()太自然了。先看一个串行基线：下面的代码调用 4 次do_some_work()，每次耗时约 1 秒：

import ray import time def do_some_work(x): time.sleep(1) # 替换为你需要做的实际工作 return x start = time.time() results = [do_some_work(x) for x in range(4)] print("duration =", time.time() - start) print("results =", results)

运行结果符合预期，总耗时约 4 秒：

duration = 4.0149290561676025 results = [0, 1, 2, 3]

常见错误一：只加装饰器，忘取结果

很多新手"并行化"的第一步就是给函数加@ray.remote并把调用改成.remote(...)：

import time import ray ray.init(num_cpus=4) # 指定系统有 4 个 CPU @ray.remote def do_some_work(x): time.sleep(1) return x start = time.time() results = [do_some_work.remote(x) for x in range(4)] print("duration =", time.time() - start) print("results =", results)

执行结果让新手一头雾水：

duration = 0.0003619194030761719 results = [ObjectRef(df5a1a828c9685d3ffffffff0100000001000000), ObjectRef(cb230a572350ff44ffffffff0100000001000000), ...]

两个关键信息：其一，程序"瞬间"完成（不到 1ms）——因为这里测的只是发起任务调用的时间，不是任务运行时间；其二，拿到的是 4 个 ObjectRef 而不是[0, 1, 2, 3]——因为远程操作异步返回 future 而非结果本身。这也印证了源码中ray.get的语义：ObjectRef 只是"结果的标识"，必须显式取回。

常见错误二：每个任务立刻 get，阻塞扼杀并行

"那就在调用后取结果呗"，于是改成：

results = [ray.get(do_some_work.remote(x)) for x in range(4)]

结果正确了，但耗时依然 4 秒，零加速：

duration = 4.018050909042358 results = [0, 1, 2, 3]

原因正如前文所说：ray.get()是阻塞调用，在每次远程调用后立刻调用它，等于等上一个任务跑完才发起下一个——本质上还是逐个执行，并行度为零。

正确姿势：先全部提交，再一次取回

要让 4 个任务真正并行，应当先批量发起所有远程调用，再统一取结果：

results = ray.get([do_some_work.remote(x) for x in range(4)])

此时耗时降到约 1 秒，说明 4 个do_some_work()确实在并行执行：

duration = 1.0064549446105957 results = [0, 1, 2, 3]

小结：ray.get()是阻塞操作，过早调用会破坏并行度；应尽量把ray.get()推迟到程序最后、把所有远程调用提交完毕之后再调用。关于"循环内反复 get"、"不必要的 get"等更细的对抗模式，仓库中的 ray-get-loop.rst、unnecessary-ray-get.rst 和 ray-get-submission-order.rst 有更深入的讨论，源码中ray.get的 docstring 也直接挂接了这些模式文档。

Tip 2：避免过小的任务，用大任务摊销调度开销

新手并行化的"自然冲动"是把每个函数、每个类都变成 remote，但任务过小反而可能让 Ray 程序比串行 Python 更慢。再次以do_some_work为例，这次把单次任务缩短到 0.1ms，并把调用次数放大到 100,000 次：

import time def tiny_work(x): time.sleep(0.0001) # 替换为你需要做的实际工作 return x start = time.time() results = [tiny_work(x) for x in range(100000)] print("duration =", time.time() - start)

串行基线约 13.4 秒：

duration = 13.36544418334961

这与理论下限吻合：10 万个 0.1ms 任务的下限是 10 秒，再加上函数调用等开销，13 秒在预期之内。现在用 Ray 并行化，让每个tiny_work()调用都变成远程任务：

import time import ray @ray.remote def tiny_work(x): time.sleep(0.0001) return x start = time.time() result_ids = [tiny_work.remote(x) for x in range(100000)] results = ray.get(result_ids) print("duration =", time.time() - start)

结果出人意料——不仅没有加速，反而更慢了：

duration = 27.46447515487671

原因在于：每一次任务调用都有不可忽略的开销（任务调度、进程间通信、系统状态更新等），当任务本身只有 0.1ms 时，调度开销完全支配了执行时间。

解法：聚合小任务，摊销单次开销

一种有效的提速方案是把远程任务"做大"，让一次性调用摊销掉启动开销。下面用mega_work把每 1000 次tiny_work()聚合进一个更大的远程函数：

import time import ray def tiny_work(x): time.sleep(0.0001) return x @ray.remote def mega_work(start, end): return [tiny_work(x) for x in range(start, end)] start = time.time() result_ids = [] [result_ids.append(mega_work.remote(x * 1000, (x + 1) * 1000)) for x in range(100)] results = ray.get(result_ids) print("duration =", time.time() - start)

运行耗时约 3.25 秒：

duration = 3.2539820671081543

大约是串行执行的 1/4，与 4 个 CPU 并行执行的预期完全吻合。

如何估算"多大的任务才算够大"

自然的问题是：任务到底多大才足以摊销远程调用开销？一个实用方法是直接测量单任务调用开销。运行下面这个"空任务"基准：

@ray.remote def no_work(x): return x start = time.time() num_calls = 1000 [ray.get(no_work.remote(x)) for x in range(num_calls)] print("per task overhead (ms) =", (time.time() - start) * 1000 / num_calls)

在 2018 款 MacBook Pro 上的实测结果为：

per task overhead (ms) = 0.4739549160003662

即执行一个空任务也要近 0.5ms。这意味着任务至少应耗时数毫秒才能摊销调用开销。当然，单任务开销因机器而异、因本机任务与跨机远程任务而异，但"让任务至少运行几毫秒"是开发 Ray 程序时一个非常实用的经验法则。这一主题的更多讨论可参考仓库中的 too-fine-grained-tasks.rst。

Tip 3：避免向远程任务重复传递同一大对象

当把一个大对象作为参数传给远程函数时，Ray 会在底层自动调用ray.put()把它存入本地对象存储。任务在本地执行时这会显著提升性能，因为所有本地任务共享同一个对象存储。但某些场景下，这种自动ray.put()反而成为性能瓶颈——典型例子就是重复传递同一个大对象：

import time import numpy as np import ray @ray.remote def no_work(a): return start = time.time() a = np.zeros((5000, 5000)) result_ids = [no_work.remote(a) for x in range(10)] results = ray.get(result_ids) print("duration =", time.time() - start)

只调用 10 个什么都不做的远程任务，却耗时约 1.08 秒：

duration = 1.0837509632110596

原因：每次调用no_work(a)，Ray 都会自动执行ray.put(a)，把数组a复制进对象存储。a有 250 万个元素（5000×5000），复制开销不容小觑，10 次调用就是 10 次全量复制。

解法：显式 ray.put 一次，传递对象 ID

避免重复复制的方法很简单：显式调用一次ray.put(a)，然后把a的 ID 传给no_work()：

import time import numpy as np import ray ray.init(num_cpus=4) @ray.remote def no_work(a): return start = time.time() a_id = ray.put(np.zeros((5000, 5000))) result_ids = [no_work.remote(a_id) for x in range(10)] results = ray.get(result_ids) print("duration =", time.time() - start)

耗时骤降至约 0.13 秒：

duration = 0.132796049118042

比原程序快了约 7 倍——因为复制数组a的操作从 10 次降为 1 次。从源码看，ray.put在 python/ray/_private/worker.py 中通过worker.put_object()落盘到对象存储并返回 ObjectRef（L3074–L3083），且对象在仍有引用期间不会被逐出（put的 docstring 明确说明："The object may not be evicted while a reference to the returned ID exists"）。

另一个隐藏收益：防止对象存储过早填满

相比提速，避免同一对象的多次拷贝还有一个更重要的好处：防止对象存储过早被填满、触发对象逐出（eviction）。频繁的逐出与重建会带来额外的传输和计算开销。仓库中的 pass-large-arg-by-value.rst 与 return-ray-put.rst 分别从"大参数按值传递"与"返回值 put 后再传"两个角度，对对象存储与引用的正确用法做了更细致的剖析。

Tip 4：流水线处理数据，谁先完成谁先处理

如果对多个任务的结果统一调用ray.get()，就必须等最慢的那个任务跑完才能开始处理。当各任务耗时差异很大时，这会造成明显的等待浪费。考虑这样一个场景：4 个do_some_work()并行执行，每个任务耗时在 0~4 秒之间均匀随机分布；随后由process_results()处理这些结果，每个结果处理 1 秒。预期总耗时 =（最慢任务的耗时）+（4 秒处理时间）。

import time import random import ray @ray.remote def do_some_work(x): time.sleep(random.uniform(0, 4)) # 替换为你需要做的实际工作 return x def process_results(results): sum = 0 for x in results: time.sleep(1) # 替换为实际处理代码 sum += x return sum start = time.time() data_list = ray.get([do_some_work.remote(x) for x in range(4)]) sum = process_results(data_list) print("duration =", time.time() - start, "\nresult = ", sum)

实测接近 8 秒：

duration = 7.82636022567749 result = 6

等最慢任务的同时，其余任务早就完成却只能干等，白白拉长了总时长。更优的做法是数据一就绪就立即处理——这正是ray.wait()的用武之地。不指定额外参数时，ray.wait()会在参数列表中任意一个对象就绪时立即返回，返回值为两个列表：(1) 已就绪对象的 ID；(2) 尚未就绪对象的 ID。把process_results()替换为每次只处理一个结果的process_incremental()：

import time import random import ray @ray.remote def do_some_work(x): time.sleep(random.uniform(0, 4)) # 替换为你需要做的实际工作 return x def process_incremental(sum, result): time.sleep(1) # 替换为实际处理代码 return sum + result start = time.time() result_ids = [do_some_work.remote(x) for x in range(4)] sum = 0 while len(result_ids): done_id, result_ids = ray.wait(result_ids) sum = process_incremental(sum, ray.get(done_id[0])) print("duration =", time.time() - start, "\nresult = ", sum)

总耗时降到约 4.85 秒，提升显著：

duration = 4.852453231811523 result = 6

两种执行方式的差异如下图所示：

图中 (a) 展示使用ray.get()等所有do_some_work()任务完成后才调用process_results()的时间线：最慢任务约在时刻 4 结束，随后 4 秒顺序处理，总计约 8 秒；(b) 展示使用ray.wait()的流水线时间线：每个任务完成即触发process_incremental()，处理与剩余任务的执行重叠，总耗时降至约 4.8 秒。

wait() 的关键参数：num_returns 与 timeout

ray.wait()的能力远不止"每轮返回一个就绪 ID"。从 python/ray/_private/worker.py 中wait的签名（L3093–L3098）可以看到两个实用参数：

• num_returns（默认 1）：指定本轮返回的就绪对象数量。例如ray.wait(result_ids, num_returns=4)会等到 4 个结果全部就绪才返回，等价于一次性的全量等待；num_returns=2则可每两个一批进行流水线处理。
• timeout（默认 None）：最多等待的秒数。设置后函数在"就绪数量达标"或"超时"两者中先到者触发返回；timeout=None表示无限等待直到满足num_returns。注意timeout必须为非负数，源码中对此有显式校验（L3191–L3194）。
• fetch_local（默认 True）：为 True 时等待对象下载到本地节点后才算就绪；为 False 时对象在集群任意位置可用即返回，不触发向本地节点的拉取。

此外，wait的输入必须是 ObjectRef（或 ObjectRefGenerator）的列表，传入单个引用会直接抛出TypeError（L3175–L3182）；返回的两个列表均保持输入顺序（L3126–L3129）。循环配合ray.wait实现"生产-消费"式流水线，是 Ray 中提升吞吐的通用范式，仓库中的 pipelining.rst 专门讨论了如何通过"提前请求下一项、再处理当前项"来让计算与 RPC 传输重叠，从而压满 CPU；该文档也指出 Ray Data 等官方库重度依赖这种流水线技术。

总结：四条性能纪律

1. 延迟ray.get()：先批量提交所有远程调用，再统一取结果；阻塞调用越晚，并行窗口越大。
2. 避免微任务：任务至少运行几毫秒，才能摊销约 0.5ms 级的单次调用开销；过小的任务应聚合为更大的远程函数。
3. 复用对象引用：同一大对象反复传参时，先ray.put()一次再传 ID，既省 7 倍时间，也避免对象存储被反复拷贝填满。
4. 流水线处理：任务耗时参差时用ray.wait()让处理紧跟就绪结果，用num_returns/timeout控制批量与超时语义，让计算与传输/等待重叠。

如果你希望进一步深入，本仓库的 ray-core 模式库 提供了二十余篇针对具体反模式的专题文章（如不必要的 get、get 循环、过细任务、大参数传值、流水线等），是这四个 Tip 的进阶版本；核心 API 的权威语义则以 python/ray/_private/worker.py 中的get、put、wait实现与 docstring 为准。

【免费下载链接】rayRay is an AI compute engine. Ray consists of a core distributed runtime and a set of AI Libraries for accelerating ML workloads.项目地址: https://gitcode.com/gh_mirrors/ra/ray