2. Ray远程函数与Actor基础

任务目标

本案例通过实现一个基于Ray Actor的计数器，掌握Ray的两类核心编程原语：远程函数（Ray Task）与Actor。你将看到远程函数如何并行执行无状态计算，以及Actor如何以有状态方式跨多次调用维护数据，并理解两者在Ray调度中的角色差异。

环境依赖

需要已安装 ray 与 pytest（仅用于验证）。安装命令：

pip install ray pytest

完整实现

以下代码定义了一个远程函数用于并行加法，以及一个计数器Actor。主程序依次演示：远程函数并行调用、Actor状态递增、以及通过Ray的 wait 获取任务完成状态。

import ray
import time

# 初始化Ray（若已存在则自动复用）
if not ray.is_initialized():
    ray.init()

# ---------- 远程函数（Task） ----------
@ray.remote
def add(a, b):
    time.sleep(0.1)  # 模拟耗时计算
    return a + b

# ---------- Actor（有状态服务） ----------
@ray.remote
class Counter:
    def __init__(self, init_value=0):
        self.value = init_value

    def increment(self, step=1):
        self.value += step
        return self.value

    def get_value(self):
        return self.value

# ---------- 主程序：任务调度演示 ----------
if __name__ == "__main__":
    # 1. 并行提交4个远程函数，不阻塞等待
    futures = [add.remote(i, i * 2) for i in range(4)]
    print("任务已提交，尚未获取结果")

    # 2. 使用wait等待全部完成（演示调度状态查询）
    done, not_done = ray.wait(futures, num_returns=4, timeout=5)
    assert len(done) == 4, f"预期4个任务完成，实际{len(done)}"
    results = ray.get(done)
    print(f"远程函数结果: {sorted(results)}")  # 预期 [0, 3, 6, 9]

    # 3. 创建Counter Actor实例
    counter = Counter.remote(10)

    # 4. 串行调用increment三次（每次+2）
    vals = []
    for _ in range(3):
        vals.append(ray.get(counter.increment.remote(2)))
    print(f"Actor计数过程: {vals}")  # 预期 [12, 14, 16]

    # 5. 验证Actor状态保持
    final_val = ray.get(counter.get_value.remote())
    print(f"Actor最终值: {final_val}")  # 预期 16
    assert final_val == 16, "Actor状态未正确保持"

    ray.shutdown()
    print("验证通过")

运行与验证

直接运行脚本：

python counter_demo.py

预期输出（严格由代码逻辑推出）：

任务已提交，尚未获取结果
远程函数结果: [0, 3, 6, 9]
Actor计数过程: [12, 14, 16]
Actor最终值: 16
验证通过

判定通过条件：程序无异常退出，且最后一行打印 验证通过。若在本地多核环境运行，可通过 ray.get_runtime_context().get_task_id() 观察远程函数实际分布在不同worker上（本案例未展开，仅说明调度机制）。

核心机制说明

• 远程函数：每次调用 add.remote() 都会创建一个独立任务，由Ray调度到集群中的可用worker执行。任务之间无共享状态，适合纯函数式并行。
• Actor：Counter.remote(10) 在集群中创建一个有状态对象，后续所有 increment.remote() 调用都路由到该Actor实例，按提交顺序串行执行，从而保证计数一致性。
• 调度差异：远程函数每次调用可能被调度到不同worker；Actor的方法调用则固定由持有该Actor的worker处理，这是两者在状态管理上的根本区别。

常见问题

• 端口冲突：若多次运行未正常shutdown，可先执行 ray.shutdown() 或重启终端。
• 结果顺序：远程函数返回顺序与提交顺序无关，示例中对结果排序仅为便于断言。
• Windows环境：需将主程序放入 if __name__ == "__main__" 保护块，否则多进程重新导入时会递归创建任务。