如何实现从Python应用触发Laravel队列任务？

下面是几种可行的实现方案，根据你的Laravel队列驱动选择对应的方式：

方法一：通过HTTP API触发队列任务 #

这是最直观、兼容性最好的方式，无需直接操作Laravel底层存储。

1. 在Laravel中创建API端点
   先定义路由指向任务分发控制器：

   // routes/api.php
   Route::post('/dispatch-task', [QueueTaskController::class, 'dispatch']);
   

   再创建控制器，在方法内分发指定队列任务：

   // app/Http/Controllers/QueueTaskController.php
   namespace App\Http\Controllers;
   
   use App\Jobs\SendNotification;
   use Illuminate\Http\Request;
   
   class QueueTaskController extends Controller
   {
       public function dispatch(Request $request)
       {
           $validated = $request->validate([
               'user_id' => 'required|integer',
               'message' => 'required|string',
               // 根据任务需求添加其他字段验证
           ]);
   
           // 分发队列任务
           SendNotification::dispatch($validated);
   
           return response()->json(['status' => 'success']);
       }
   }
   

2. Python端发送请求
   使用requests库向API端点发送POST请求，携带任务参数：

   import requests
   
   def trigger_laravel_queue():
       url = "https://your-laravel-app.com/api/dispatch-task"
       payload = {
           "user_id": 123,
           "message": "Hello from Python!"
       }
       headers = {
           "Accept": "application/json",
           "Authorization": "Bearer YOUR_API_TOKEN"  # 若Laravel启用API认证需添加
       }
   
       response = requests.post(url, json=payload, headers=headers)
       if response.status_code == 200:
           print("任务触发成功")
       else:
           print(f"触发失败: {response.text}")
   
   trigger_laravel_queue()
   

方法二：直接操作Laravel数据库队列表（适用于数据库驱动） #

如果Laravel队列使用数据库驱动，可直接在Python中往jobs表插入任务数据。

1. 核心字段说明
   jobs表关键字段：

   • queue: 队列名称（默认default）
   • payload: 任务序列化数据，需匹配Laravel格式
   • available_at: 任务可执行时间戳（立即执行填当前时间戳）
   • created_at: 任务创建时间戳

2. Python端插入任务
   以MySQL为例，使用pymysql库操作：

   import pymysql
   import json
   import time
   import uuid
   import base64
   
   def insert_db_queue_task():
       conn = pymysql.connect(
           host='your-db-host',
           user='db-user',
           password='db-password',
           database='laravel-db-name'
       )
       cursor = conn.cursor()
   
       # 构造符合Laravel格式的payload
       command_data = json.dumps({
           "user_id": 123,
           "message": "Hello from Python DB",
           "connection": None,
           "queue": None,
           "chainConnection": None,
           "chainQueue": None,
           "chainCatchCallbacks": None,
           "delay": None,
           "middleware": [],
           "chained": []
       }).encode('utf-8')
   
       payload = json.dumps({
           "uuid": str(uuid.uuid4()),
           "displayName": "App\\Jobs\\SendNotification",
           "job": "Illuminate\\Queue\\CallQueuedHandler@call",
           "maxTries": None,
           "maxExceptions": None,
           "failOnTimeout": False,
           "backoff": None,
           "timeout": None,
           "timeoutAt": None,
           "data": {
               "commandName": "App\\Jobs\\SendNotification",
               "command": base64.b64encode(command_data).decode('utf-8')
           }
       })
   
       current_ts = int(time.time())
       sql = """
           INSERT INTO jobs (queue, payload, available_at, created_at)
           VALUES (%s, %s, %s, %s)
       """
       cursor.execute(sql, ('default', payload, current_ts, current_ts))
       conn.commit()
   
       cursor.close()
       conn.close()
       print("数据库队列任务已插入")
   
   insert_db_queue_task()
   

方法三：直接推送任务到Redis队列（适用于Redis驱动） #

若Laravel使用Redis队列，可直接用Python往对应Redis队列Key推送任务。

import redis
import json
import uuid
import base64

def push_redis_queue_task():
    r = redis.Redis(
        host='your-redis-host',
        port=6379,
        password='redis-password',
        db=0
    )

    # 构造任务数据
    task_data = {
        "user_id": 123,
        "message": "Hello from Python Redis"
    }

    command_data = json.dumps({
        **task_data,
        "connection": None,
        "queue": None,
        "chainConnection": None,
        "chainQueue": None,
        "chainCatchCallbacks": None,
        "delay": None,
        "middleware": [],
        "chained": []
    }).encode('utf-8')

    payload = {
        "uuid": str(uuid.uuid4()),
        "displayName": "App\\Jobs\\SendNotification",
        "job": "Illuminate\\Queue\\CallQueuedHandler@call",
        "maxTries": None,
        "maxExceptions": None,
        "failOnTimeout": False,
        "backoff": None,
        "timeout": None,
        "timeoutAt": None,
        "data": {
            "commandName": "App\\Jobs\\SendNotification",
            "command": base64.b64encode(command_data).decode('utf-8')
        }
    }

    # 往默认队列推送，Redis队列Key格式为"queues:队列名"
    r.lpush('queues:default', json.dumps(payload))
    print("Redis队列任务已推送")

push_redis_queue_task()

注意事项 #

• 确保Python构造的payload格式与Laravel生成的完全一致，尤其是任务类命名空间、command的base64编码部分。
• 若需延迟执行任务，数据库驱动修改available_at为延迟后的时间戳；Redis驱动则将任务推送到延迟队列（如queues:default:delayed），用zadd设置score为延迟时间戳。
• 生产环境优先选择HTTP API方式，更安全易维护，避免直接操作底层存储带来的兼容性问题。

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