Ray python 分布式计算

import rayimport timeimport iofrom PIL import Image# =====================================================================# 步骤 1：连接集群# =====================================================================# 提示：如果是单机本地测试，直接写 ray.init() 即可，Ray 会在本地拉起虚拟集群。# 如果是在多机集群上运行，填入 "auto"，Ray 会自动连接你在命令行启动的 6379 端口。ray.init(address="auto" if ray.is_initialized() else None)print("Ray 分布式集群连接成功！开始执行任务...")# =====================================================================# 步骤 2：定义【有状态的 Actor】 —— 全局计数器# =====================================================================@ray.remoteclass ClusterMetricsCounter:    def __init__(self):        self.total_processed = 0        self.node_breakdown = {}    def log_success(self, node_ip: str):        """记录处理成功事件，并累加当前节点的计数"""        self.total_processed += 1        self.node_breakdown[node_ip] = self.node_breakdown.get(node_ip, 0) + 1        return self.total_processed    def fetch_report(self):        """返回当前的全局统计报表"""        return {            "total_images": self.total_processed,            "distribution": self.node_breakdown        }# =====================================================================# 步骤 3：定义【无状态的 Task】 —— 图像分布式转换# =====================================================================@ray.remotedef process_image_task(image_bytes: bytes, metrics_actor_handle) -> dict:    """    接收图片字节流，在远程节点将其转换为灰度图。    处理完成后，异步向集群中的有状态 Actor 汇报状态。    """    # 动态获取当前 Task 到底被分配到了哪台物理机器的 IP 上    current_node_ip = ray.get_runtime_context().worker.node_ip_address    # 模拟图像处理逻辑    img = Image.open(io.BytesIO(image_bytes))    gray_img = img.convert("L")     # 故意休眠 1 秒，模拟复杂的重度计算开销    time.sleep(1)     # 【高能交互】：Task 跨网络远程调用 Actor 的方法，递增全局计数    # 注意：这里的 .remote() 是非阻塞的，Task 报完信会立刻往下走    metrics_actor_handle.log_success.remote(current_node_ip)    return {        "status": "success",        "processed_by_ip": current_node_ip,        "output_size": gray_img.size    }# =====================================================================# 步骤 4：主控制流执行（Driver 进程）# =====================================================================if __name__ == "__main__":    start_time = time.time()    # A. 在集群某处实例化常驻的 Actor    print("\n[1/4] 正在初始化远程有状态 Actor...")    metrics_actor = ClusterMetricsCounter.remote()    # B. 准备模拟数据（生成 6 张红色空白图片的二进制流）    print("[2/4] 准备图片测试数据...")    mock_img = Image.new("RGB", (800, 800), color="red")    img_byte_arr = io.BytesIO()    mock_img.save(img_byte_arr, format='JPEG')    raw_bytes = img_byte_arr.getvalue()    # C. 异步触发 6 个分布式任务    print("[3/4] 开始跨集群异步分发 6 个无状态图像处理任务...")    # 我们把刚才创建的 metrics_actor 句柄像普通变量一样作为参数传进去    futures = [process_image_task.remote(raw_bytes, metrics_actor) for _ in range(6)]    print("提示：.remote() 会立即返回 Future（对象引用），主线程完全没有被阻塞！")    # D. 阻塞等待，回收全集群的计算结果    print("\n[4/4] 正在通过 ray.get() 同步阻塞等待所有远程节点计算返回...")    task_results = ray.get(futures)    # E. 打印任务明细与 Actor 收集的全局状态    print("\n" + "="*30 + " 集群物理执行明细 " + "="*30)    for idx, res in enumerate(task_results):        print(f"图片 {idx+1} 处理完毕 | 实际运行物理机器 IP: {res['processed_by_ip']} | 尺寸: {res['output_size']}")    # 远程调用 Actor 获取它帮我们保管的内存状态    final_report = ray.get(metrics_actor.fetch_report.remote())    print("\n" + "="*30 + " 远程 Actor 全局账本报表 " + "="*30)    print(f"整个集群累计处理图片总数: {final_report['total_images']}")    print(f"各台服务器工作量负载打散详情: {final_report['distribution']}")    print("="*75)    print(f"整个流程总耗时: {time.time() - start_time:.2f} 秒。")    print("（由于分布式并行，6个耗时1秒的任务在多核/多机上并发执行，总时间远小于 6 秒！）")    # 关闭 Ray 连接    ray.shutdown()