本文介绍一种基于 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 条✅ 关键优势说明：
- 零共享状态：无需 threading.Queue、ray.util.queue.Queue 或外部 Redis，规避跨进程/跨节点序列化与锁竞争；
- 天然隔离：每个 Actor 拥有独立 data_buffer，无并发修改风险；
- 低延迟读取：pop_data() 始终可立即响应，不受 generate() 循环影响；
- 弹性伸缩：Actor 数量与 max_concurrency 可按负载独立调整。
⚠️ 注意事项：
- max_concurrency 必须 ≥ 2，否则 pop_data() 调用将永远等待 generate() 结束（而它永不结束）；
- random.random() 替代了原文误写的 random.rand()（后者属于 NumPy，需 import numpy as np）；
- 若需强一致性（如严格 FIFO 全局顺序），本方案不适用——此时应引入中心化队列（如 ray.util.queue.Queue）并配合 ray.wait() 流控；
- 缓冲区过大会增加内存压力，建议在 pop_data() 中做截断或在主线程中定期清理。
该模式是 Ray 官方推荐的“Actor 内部状态 + 并发访问”典型范式，兼顾简洁性、性能与可维护性，适用于实时数据采集、日志聚合、传感器流预处理等场景。
