Ray框架入门：Ray Core核心概念与并行计算实战

Ray 是当前分布式强化学习（DRL）训练中最常用的底层计算框架之一，它通过简洁的 Python API 将单机代码扩展到集群，同时保留了近乎原生的开发体验。本章将围绕 Ray Core 的三大核心抽象——Tasks（任务）、Actors（状态化执行体）和 Object Store（分布式对象存储）展开，并结合安装配置、基本 API 使用与并行计算示例，帮助你建立对 Ray 运行机制的完整认知。理解这些概念是后续实现 PPO、DQN 等分布式训练算法的前提。

Ray Core 核心抽象与数据流 Driver (主进程) Python 脚本入口 Object Store (分布式对象存储) 共享内存 / Plasma / 零拷贝 Worker 1 执行 Task / Actor Worker 2 执行 Task / Actor Worker 3 执行 Task / Actor Actor (有状态) Replay Buffer Network Parameters Env State / Metrics put / get 零拷贝读取 返回 ObjectRef Actor 创建与远程方法调用 核心数据流说明 Driver 提交任务并获取 ObjectRef（未来引用） Task 无状态执行，结果写入 Object Store Actor 维护状态，跨方法调用保持数据

上图展示了 Ray 运行时的核心数据流：Driver 进程负责提交任务，Task 在无状态 Worker 中执行，Actor 则维护跨调用状态，所有中间结果通过 Object Store 传递。下面我们逐一深入这些概念。

3.1 Ray Core 三大核心概念

3.1.1 Tasks：无状态并行任务

Task 是 Ray 中最基础的并行执行单元。通过 @ray.remote 装饰器，一个普通 Python 函数可以转换为远程任务。每次调用该函数时，Ray 调度器会将其分发到集群中某个可用 Worker 上执行。Task 的关键特征是无状态性——每次执行都是独立的，不保留任何跨调用的内存数据。

核心原理：Task 的调度由 Ray 的中央调度器（GCS，Global Control Store）负责。当 Driver 调用远程函数时，Ray 会创建任务规格（Task Spec），包含函数 ID、参数引用和资源需求。调度器根据集群中每个节点的可用资源（CPU、GPU、内存）进行匹配，将任务分配到最合适的 Worker 上。任务执行完成后，返回值被序列化并存入 Object Store，Driver 通过 ObjectRef 异步获取结果。

import ray
import time

ray.init()

@ray.remote
def compute_square(x):
    time.sleep(1)  # 模拟计算
    return x * x

# 异步提交 4 个任务
futures = [compute_square.remote(i) for i in range(4)]
# 阻塞获取结果
results = ray.get(futures)
print(results)  # [0, 1, 4, 9]

上述代码中，compute_square.remote(i) 立即返回一个 ObjectRef，而不会阻塞主线程。4 个任务可以并行执行，总耗时约 1 秒而非 4 秒。这种异步模式是 Ray 实现高吞吐的关键。

3.1.2 Actors：有状态的服务实体

强化学习训练中，经验回放缓冲区（Replay Buffer）、环境模拟器、策略网络参数等都需要跨步骤维护状态。Ray 的 Actor 抽象正是为此设计。Actor 是一个有状态的 Python 类实例，运行在独立的 Worker 进程中，通过远程方法调用来修改其内部状态。

@ray.remote
class ReplayBuffer:
    def __init__(self, capacity):
        self.capacity = capacity
        self.data = []
    
    def push(self, item):
        if len(self.data) < self.capacity:
            self.data.append(item)
        else:
            self.data.pop(0)
            self.data.append(item)
    
    def sample(self, batch_size):
        import random
        return random.sample(self.data, batch_size)

# 创建 Actor 实例
buffer = ReplayBuffer.remote(capacity=1000)
# 远程调用方法
buffer.push.remote((1, 2, 3))
buffer.push.remote((4, 5, 6))
samples = ray.get(buffer.sample.remote(1))
print(samples)  # [(1, 2, 3)] 或 [(4, 5, 6)]

Actor 的每个方法调用默认是串行执行的（同一 Actor 内部），这保证了状态的一致性。在分布式 RL 中，通常每个环境模拟器对应一个 Actor，多个 Actor 并行推进环境步进，而全局模型参数则存储在另一个共享 Actor 中。

3.1.3 Object Store：分布式内存对象存储

Object Store 是 Ray 的数据传输中枢。所有 Task 的输入参数和返回值、Actor 方法的参数和结果，都通过 Object Store 传递。Ray 使用 Apache Arrow 作为内存数据格式，支持零拷贝共享，极大减少了序列化开销。

特性说明RL 训练中的典型应用
存储位置每个节点本地共享内存 + 分布式元数据经验轨迹暂存
数据格式Apache Arrow 列式格式批量状态、动作、奖励张量
访问方式ObjectRef 引用，ray.get() 获取从 Replay Buffer 采样
生命周期引用计数自动回收训练循环中的临时结果
共享机制同一节点内零拷贝，跨节点序列化传输参数同步

经验提醒：在 DRL 训练中，尽量避免将大体积的 numpy 数组频繁通过 ray.get() 拉回 Driver。更好的做法是让数据在 Worker 之间直接传递，或者使用 ray.put() 将大对象预先放入 Object Store，然后传递 ObjectRef 而非对象本身。

3.2 安装配置与集群启动

3.2.1 单机安装

Ray 支持 pip 直接安装，推荐使用 Python 3.8-3.11 版本。安装命令如下：

# 基础安装（CPU 版本）
pip install ray

# 安装强化学习相关依赖
pip install "ray[rllib]" gymnasium

# 如果使用 GPU 训练，确保 CUDA 环境正确
pip install ray[default]

安装完成后，可以通过 ray.init() 初始化本地集群。默认情况下，Ray 会占用当前机器的所有可用 CPU 核心。可以通过参数限制资源使用：

ray.init(num_cpus=4, num_gpus=0, include_dashboard=True)
# 输出显示 Dashboard 地址，例如 http://127.0.0.1:8265

3.2.2 集群部署

对于多机训练，需要先启动 Head 节点，然后让 Worker 节点加入。Ray 集群的部署方式有两种：手动启动和通过集群启动器（Cluster Launcher）。

# 在 Head 节点上
ray start --head --port=6379 --dashboard-host=0.0.0.0

# 在 Worker 节点上（替换为 Head 节点 IP）
ray start --address=192.168.1.10:6379

# 在 Python 中连接集群
ray.init(address="auto")

风险提示：集群模式下，所有节点必须能够通过 SSH 互相访问，且防火墙需要开放 6379（GCS）、8265（Dashboard）和 10001-10010（对象存储）端口。如果使用云厂商的机器，还需要配置安全组规则。

3.3 基本 API 使用与并行计算示例

3.3.1 核心 API 速查

API功能示例
ray.init()初始化 Ray 运行时ray.init(address="auto")
@ray.remote将函数/类转换为远程对象@ray.remote def f(x)
.remote()异步执行远程函数/方法f.remote(1)
ray.get()阻塞获取 ObjectRef 的值ray.get(ref)
ray.wait()等待部分任务完成ray.wait([ref1, ref2], num_returns=1)
ray.put()将对象放入 Object Storeref = ray.put(large_data)
ray.cancel()取消正在执行的任务ray.cancel(ref)

3.3.2 并行计算实战：批量策略评估

假设我们有一个简单的策略网络（用随机权重模拟），需要并行评估 100 个不同初始状态下的动作价值。这个场景非常适合用 Task 并行化。

import numpy as np
import ray

ray.init(num_cpus=4)

# 模拟策略网络的前向计算
@ray.remote
def evaluate_policy(state, weights):
    # 模拟计算 Q 值
    q_values = np.dot(state, weights)
    return np.max(q_values)

# 生成模拟数据
states = [np.random.randn(64) for _ in range(100)]
shared_weights = np.random.randn(64, 4)

# 将共享权重放入 Object Store，避免重复传输
weights_ref = ray.put(shared_weights)

# 并行提交 100 个评估任务
futures = [evaluate_policy.remote(s, weights_ref) for s in states]

# 分批获取结果，避免一次性加载过多
results = []
for i in range(0, 100, 10):
    batch = ray.get(futures[i:i+10])
    results.extend(batch)

print(f"评估完成，平均 Q 值: {np.mean(results):.4f}")

这个示例展示了三个关键技巧：共享大对象预放置（ray.put）、批量异步提交、分块获取结果。在实际 RL 训练中，这种模式常用于并行采样多个环境的经验数据。

3.4 常见误区与总结复盘

3.4.1 误区一：过度使用 ray.get() 导致阻塞

初学者常犯的错误是在循环中频繁调用 ray.get()，这会破坏并行性。正确做法是先将所有任务提交，最后统一获取结果。

反例：

# 错误：每提交一个任务就等待，串行执行
for i in range(10):
    ref = f.remote(i)
    result = ray.get(ref)  # 阻塞等待，失去并行性

正解：

# 正确：先全部提交，再统一获取
refs = [f.remote(i) for i in range(10)]
results = ray.get(refs)

3.4.2 误区二：忽略 Actor 的并发限制

同一 Actor 的方法调用是串行的。如果多个 Driver 线程同时调用同一个 Actor 的方法，它们会排队执行。在 DRL 中，如果所有环境都放在一个 Actor 中，训练速度会严重受限。解决方案是创建多个 Actor 实例，每个管理少量环境。

3.4.3 误区三：对象存储内存溢出

Object Store 默认使用节点可用内存的 30% 作为上限。如果大量对象未被引用且未释放，可能导致内存溢出。Ray 使用引用计数自动回收，但循环引用或长期持有 ObjectRef 会导致内存泄漏。建议使用 ray.get() 后及时删除引用，或使用 del 释放。

复盘总结：Ray Core 的设计哲学是“用最小的 API 改动实现最大的并行度”。掌握 Task 的无状态并行、Actor 的有状态服务、Object Store 的高效数据共享，是构建分布式 RL 训练系统的基石。在后续章节中，我们将基于这些原语实现完整的 PPO 训练器。

通过本章的学习，你应该能够独立完成 Ray 的安装配置，理解 Tasks、Actors、Object Store 的适用场景，并能够编写基本的并行计算代码。下一阶段，我们将深入 Ray 的调度策略和资源管理，为大规模训练做准备。

← 上一章 📖 返回目录 下一章 →