如何通过Python ray的分布式计算Python API实现高效并行任务处理

ray的分布式计算Python API设计简洁，开发者无需深入了解分布式底层原理，就能通过简单的装饰器和函数调用实现任务的分布式调度，适合处理机器学习训练、大规模数据计算等场景。

ray核心API基础概念

使用ray的分布式API前需要先了解几个核心概念，这些概念是理解后续API用法的基础。

• 远程函数：通过@ray.remote装饰器修饰的普通Python函数，调用时不会立即执行，而是提交到ray集群的任务队列中异步执行。
• 对象引用：远程函数调用后会返回一个对象引用，类似一个未来结果的凭证，需要通过ray.get方法才能获取实际执行结果。
• actor：有状态的可调用对象，通过@ray.remote装饰类得到，适合需要维护状态的分布式场景。

环境初始化API

使用ray的所有分布式API前，都需要先初始化ray运行时，初始化时会自动连接本地或远程的ray集群。

import ray

# 初始化ray，默认连接本地集群，若需要连接远程集群可传入address参数
# 比如连接远程集群可写ray.init(address="ray://127.0.0.1:10001")
ray.init()

远程函数定义与调用API

定义远程函数是使用ray分布式计算的第一步，通过@ray.remote装饰器即可将普通函数转换为可分布式执行的远程函数。

定义远程函数

import ray
import time

ray.init()

# 使用@ray.remote装饰器定义远程函数
@ray.remote
def add(a, b):
    # 模拟耗时计算
    time.sleep(1)
    return a + b

提交任务与获取结果

远程函数调用后会立即返回一个对象引用，不会阻塞当前线程，所有任务提交完成后可以统一调用ray.get获取结果。

# 提交3个任务，此时任务进入ray的任务队列，不会立即执行
task1 = add.remote(1, 2)
task2 = add.remote(3, 4)
task3 = add.remote(5, 6)

# 获取所有任务的结果，ray.get会阻塞直到所有结果返回
results = ray.get([task1, task2, task3])
print(results)  # 输出[3,7,11]

如果直接调用普通函数，3个任务串行执行需要3秒，而使用ray分布式执行，3个任务并行处理只需要1秒左右，效率提升明显。

资源限制API

如果某些任务需要占用较多CPU或GPU资源，可以通过num_cpus或num_gpus参数指定远程函数执行时需要的资源，ray会按照资源限制调度任务。

import ray

ray.init()

# 指定该远程函数执行需要占用2个CPU核心
@ray.remote(num_cpus=2)
def heavy_compute(x):
    # 模拟占用CPU的计算任务
    total = 0
    for i in range(x):
        total += i
    return total

# 提交任务，ray会确保执行该任务时分配2个CPU核心
task = heavy_compute.remote(1000000)
result = ray.get(task)
print(result)

Actor API用法

当任务需要维护状态时，普通远程函数无法满足需求，此时可以使用ray的Actor API，Actor是有状态的对象，每个Actor实例会独立运行在一个worker进程中。

import ray

ray.init()

# 使用@ray.remote装饰类定义Actor
@ray.remote
class Counter:
    def __init__(self):
        self.count = 0

    def increment(self):
        self.count += 1
        return self.count

    def get_count(self):
        return self.count

# 创建Actor实例，创建过程也是异步的，返回Actor的引用
counter = Counter.remote()

# 调用Actor的方法，同样返回对象引用
task1 = counter.increment.remote()
task2 = counter.increment.remote()

# 获取结果
print(ray.get(task1))  # 输出1
print(ray.get(task2))  # 输出2
print(ray.get(counter.get_count.remote()))  # 输出2

等待任务完成API

如果提交了大量任务，不需要等待所有任务完成再处理，可以使用ray.waitAPI等待部分任务完成，实现流水线处理。

import ray
import time

ray.init()

@ray.remote
def slow_task(x):
    time.sleep(x)
    return x

# 提交5个任务，分别耗时1-5秒
tasks = [slow_task.remote(i) for i in range(1, 6)]

# 等待至少2个任务完成，超时时间设为3秒
ready, remaining = ray.wait(tasks, num_returns=2, timeout=3)

print("已完成的任务结果：", ray.get(ready))  # 输出[1,2]
print("未完成的任务数量：", len(remaining))  # 输出3

关闭ray运行时

所有任务执行完成后，可以调用ray.shutdown关闭ray运行时，释放相关资源。

import ray

ray.init()
# 执行相关分布式任务
# ...
# 关闭ray
ray.shutdown()

常见使用注意事项

• 远程函数的参数和返回值必须是可序列化的，因为需要在不同进程之间传递，默认使用pickle序列化。
• 不要在远程函数内部调用ray.init，初始化只需要执行一次即可。
• 如果任务量很小，分布式调度的开销可能会抵消并行带来的收益，此时不建议使用ray分布式API。
• 远程函数内部无法直接访问主进程的全局变量，需要显式传递参数。

Pythonray分布式计算parallel_taskremote_function修改时间：2026-07-13 00:45:31