L4 分布式：Ray 与 Celery

目标

• 用 Ray 把 Python 函数和类变成分布式任务；
• 用 Ray Serve 部署一个可以扩缩容的服务；
• 用 Celery + Redis 搭一个任务队列，理解“生产者—队列—消费者”模式。

背景

边缘智能系统由多台异构设备组成，任务需要分发到不同机器上执行。两种常见方式：

• Ray：UC Berkeley RISELab 开发的通用分布式计算框架，用几行注解就能把函数（task）和有状态的类（actor）放到集群上运行，内置对象存储和资源调度。vLLM 的多机推理、很多强化学习和大模型训练框架都基于 Ray。
• Celery + Redis：经典的分布式任务队列。生产者把任务放进 Redis，多个 worker 从队列里取任务执行。结构简单，适合把推理请求分发给多个 worker。

环境

• Python 包：uv pip install "ray[serve]" celery redis
• Docker（运行 Redis）

第 1 步：Ray 任务与 Actor

# ray_basics.py
import time

import ray

ray.init()   # 单机运行；连接已有集群用 ray.init(address="auto")


@ray.remote
def slow_square(x):
    time.sleep(1)
    return x * x


start = time.time()
futures = [slow_square.remote(i) for i in range(8)]   # 立即返回，任务并行执行
print(ray.get(futures))                                # 等待结果
print(f"用时 {time.time() - start:.1f} 秒")            # 远小于串行的 8 秒


@ray.remote
class Counter:
    """Actor：有状态的远程对象，例如常驻显存的模型"""

    def __init__(self):
        self.n = 0

    def incr(self):
        self.n += 1
        return self.n


counter = Counter.remote()
print(ray.get([counter.incr.remote() for _ in range(3)]))   # [1, 2, 3]

资源声明：@ray.remote(num_gpus=1) 表示这个任务需要一张 GPU，Ray 会把它调度到有空闲 GPU 的节点上。这就是一个简单的资源调度器。

运行时打开 Ray Dashboard（默认 http://127.0.0.1:8265），可以看到任务、actor 和资源使用情况。

第 2 步：多机集群

# 在主节点
ray start --head --port=6379

# 在其他节点（填主节点的 IP）
ray start --address=主节点IP:6379

# 查看集群状态
ray status

之后在任意节点上运行 ray.init(address="auto") 的程序，任务就会分布到整个集群。

第 3 步：Ray Serve 部署服务

# serve_app.py
from ray import serve
from starlette.requests import Request


@serve.deployment(num_replicas=2)
class Echo:
    async def __call__(self, request: Request):
        body = await request.json()
        return {"echo": body["text"]}


serve.run(Echo.bind())
input("服务已启动，按回车退出\n")

python serve_app.py
# 另开一个终端
curl -X POST http://127.0.0.1:8000/ -d '{"text": "hello"}'

Ray Serve 默认在 8000 端口提供 HTTP 服务。如果同一台机器上还运行着 L1、L3 的 vLLM 容器（也占用 8000 端口），serve.run 会因为端口被占用而失败：先 docker stop vllm-qwen，或者在 serve.run 之前加一行 serve.start(http_options={"port": 8001})，curl 的地址也改成 8001。

num_replicas 控制副本数。把 Echo 换成加载模型、调用推理的类，就是一个最简单的多副本推理服务。研究问题：请求量变化时，副本数怎么自动伸缩？不同副本放在哪些机器上？

第 4 步：Celery + Redis 任务队列

启动 Redis：

docker run -d --name redis -p 6379:6379 redis:7

如果同一台机器上已经运行了 Ray 主节点（也占用 6379 端口），把 Redis 映射到别的端口，例如 -p 16379:6379，并修改下面代码里的地址。

定义任务：

# tasks.py
import time

from celery import Celery

app = Celery("tasks", broker="redis://localhost:6379/0", backend="redis://localhost:6379/1")


@app.task
def infer(prompt):
    time.sleep(2)   # 换成真正的模型调用
    return f"回答：{prompt}"

启动 worker（--concurrency 是并发执行的任务数）：

celery -A tasks worker --concurrency=4 --loglevel=info

提交任务：

from tasks import infer

results = [infer.delay(f"问题 {i}") for i in range(8)]
print([r.get(timeout=60) for r in results])

在另一台机器上用相同的代码启动 worker（Redis 地址改成那台机器能访问的地址），任务就会被两台机器分担。

Ray 与 Celery 怎么选

	Ray	Celery
抽象	任务、actor、对象存储、资源感知调度	任务队列
GPU 资源调度	原生支持	需要自己管理
生态	训练、调参、推理服务、强化学习	Web 后端异步任务
适合	AI 工作负载、多智能体系统	简单的任务分发

学习资料

• Ray 官方文档和 GitHub
• 视频：Ray 分布式计算的通用框架、Ray：AI 时代的统一分布式计算架构最新介绍
• 用 Ray 构建多智能体系统的思路：知乎文章