如何让多个 Ray Actor 并发写入共享数据缓冲区并由主线程安全读取-Python教程

本文介绍一种基于 ray 线程化 actor（`max_concurrency`）的轻量级方案，使多个远程 `datagenerator` 实例持续生成数据并存入各自缓冲区，主线程可无阻塞地批量拉取所有缓冲数据，避免传统共享队列的复杂性与同步风险。

在分布式数据生成场景中，常需多个 Ray Actor 并行生产数据，并由主驱动线程统一收集处理。但默认情况下，Ray Actor 是单线程串行执行的：一旦某个方法（如 generate()）进入无限循环，其他方法（如 pop_data()）将被永久阻塞，导致主线程无法及时获取数据。

核心解法：启用线程化 Actor（Threaded Actor）
通过 .options(max_concurrency=N) 显式允许 Actor 同时处理多个远程调用，即可实现“后台持续生成 + 前台即时读取”的并发模型。只需设置 max_concurrency=2，即允许多至 2 个方法并行执行——一个运行 generate() 循环，另一个响应 pop_data() 请求。

以下是完整、可直接运行的实践代码：

import ray
import random
import time

ray.init(ignore_reinit_error=True)

@ray.remote
class DataGenerator:
    def __init__(self):
        self.data_buffer = []

    def generate(self):
        while True:
            time.sleep(5)  # 模拟耗时数据生成
            data = random.random()  # 注意：原示例中 random.rand() 应为 random.random()
            self.data_buffer.append(data)

    def pop_data(self):
        """原子性获取并清空当前缓冲区"""
        data = self.data_buffer.copy()  # 避免引用共享问题
        self.data_buffer.clear()
        return data

# 启动 10 个并发 Actor，每个支持最多 2 个并发方法调用
N_HANDLES = 10
generator_handles = [
    DataGenerator.remote().options(max_concurrency=2)
    for _ in range(N_HANDLES)
]

# 启动所有生成器（非阻塞）
for handle in generator_handles:
    handle.generate.remote()

# 主线程：周期性拉取全部数据并处理
all_data = []
while True:
    # 并发获取所有 Actor 的当前缓冲数据
    results = ray.get([h.pop_data.remote() for h in generator_handles])

    # 合并数据
    for batch in results:
        all_data.extend(batch)

    print(f"已累积 {len(all_data)} 条数据")

    # ✅ 此处可执行任意计算密集型任务（如模型推理、批处理等）
    # 即使耗时较长，也不会阻塞 Actor 的数据生成
    # time.sleep(30)  # 示例：模拟长耗时处理

    # （可选）重置或限流：防止 all_data 无限增长
    if len(all_data) > 1000:
        all_data = all_data[-500:]  # 保留最新 500 条

✅ 关键优势说明：