如何在PHP中从Celery Redis Backend获取任务结果？针对gjedeer/celery-php不支持Redis后端的解决方案咨询

我之前在项目里也碰到过gjedeer/celery-php不支持Redis后端的问题，折腾了几个亲测可行的方案，给你参考：

方案1：直接操作Redis查询结果 #

Celery用Redis存储任务结果时，键的格式是固定的：celery-task-meta-<你的任务ID>，存储内容是标准JSON格式。我们完全可以绕过celery-php库，直接用PHP的Redis扩展读取解析：

实现步骤： #

1. 确保PHP已安装Redis扩展（可通过pecl install redis或系统包管理器安装）
2. 编写查询代码：

<?php
$taskId = "你的Celery任务ID";
$redisKey = "celery-task-meta-" . $taskId;

// 连接Redis服务
$redis = new Redis();
$redis->connect('localhost', 6379);
// 若Redis设置了密码，添加此行：$redis->auth('your_redis_password');

// 获取任务元数据
$taskMeta = $redis->get($redisKey);
if (!$taskMeta) {
    echo "任务不存在或尚未开始执行";
    exit;
}

// 解析JSON格式结果
$result = json_decode($taskMeta, true);

// 根据任务状态处理结果
switch ($result['status']) {
    case 'PENDING':
        echo "任务正在等待执行";
        break;
    case 'SUCCESS':
        echo "任务执行成功，结果：" . print_r($result['result'], true);
        break;
    case 'FAILURE':
        echo "任务执行失败，错误信息：" . $result['traceback'];
        break;
    default:
        echo "任务当前状态：" . $result['status'];
}
?>

这个方案最直接，无额外依赖，性能也最优。

方案2：搭建Python中间层代理 #

如果不想直接操作Redis（比如担心Celery存储格式变动、权限限制等），可以写一个轻量的Python服务，用官方Celery库获取结果，再通过HTTP接口提供给PHP调用：

示例（用Flask实现中间接口）： #

from flask import Flask, jsonify
from celery import Celery

app = Flask(__name__)
# 配置Celery，与你的Celery Worker配置保持一致
celery = Celery('tasks', backend='redis://localhost:6379/0', broker='redis://localhost:6379/0')

@app.route('/task-result/<task_id>')
def get_task_result(task_id):
    result = celery.AsyncResult(task_id)
    response = {
        'task_id': task_id,
        'status': result.status,
        'result': result.result if result.status == 'SUCCESS' else None,
        'error': result.traceback if result.status == 'FAILURE' else None
    }
    return jsonify(response)

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

PHP端请求代码： #

<?php
$taskId = "你的Celery任务ID";
$apiUrl = "http://localhost:5000/task-result/" . $taskId;

$ch = curl_init($apiUrl);
curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
$response = curl_exec($ch);
curl_close($ch);

$result = json_decode($response, true);
print_r($result);
?>

这个方案兼容性最强，后续Celery后端变更时，只需调整Python中间层，PHP端无需修改。

方案3：临时切换支持的后端（备选） #

如果项目架构允许，可以将Celery的结果后端换成gjedeer/celery-php支持的类型（比如RabbitMQ），这样就能直接用库自带方法获取结果：

<?php
require_once 'vendor/autoload.php';

$celery = new Celery('localhost', 'guest', 'guest', '/', 5672);
$result = $celery->result('你的任务ID');

echo "任务状态：" . $result['status'] . "\n";
if ($result['status'] == 'SUCCESS') {
    echo "执行结果：" . print_r($result['result'], true);
}
?>

不过此方案需要调整Celery全局配置，可能影响现有业务，优先级低于前两个方案。

内容的提问来源于stack exchange，提问作者Ali