Mastering ThinkPHP’s Think Queue: Installation, Usage, and Command Guide

Overview

Think Queue is the official message‑queue extension for the ThinkPHP framework. It is designed for high‑concurrency scenarios such as SMS, email, or app push notifications that require batch processing or long‑running tasks.

Core Features

Publish, fetch, execute, delete, retry, failure handling, delayed execution and timeout control for messages.

Support for multiple queues, memory limits, start/stop control and daemon management.

Ability to downgrade to synchronous execution when needed.

Installation

composer require topthink/think-queue

Typical Usage Flow

Consume and delete messages.

Create and push messages.

Process tasks.

Publish tasks via an API endpoint.

Consume and Delete Messages

Create a consumer class that implements a fire method. The method receives a Job instance and the custom payload.

<?php
declare(strict_types=1);

namespace app\job;

use think\facade\Log;
use think\queue\Job;

class TinywanJob
{
    /**
     * The method invoked by the queue worker.
     *
     * @param Job   $job  Current job object.
     * @param mixed $data Payload supplied when the job was pushed.
     */
    public function fire(Job $job, $data)
    {
        // If the job is no longer needed, delete it.
        if (! $this->checkJob($data)) {
            $job->delete();
            return;
        }

        // Execute business logic.
        if ($this->doJob($data)) {
            $job->delete();
            Log::info('Job completed and deleted');
        } else {
            // Retry limit: 3 attempts.
            if ($job->attempts() > 3) {
                Log::info('Job exceeded retry limit');
                $job->delete();
            }
        }
    }

    /**
     * Determine whether the job should be processed.
     */
    private function checkJob($data): bool
    {
        // Example checks (timestamp, business id, parameters)
        $ts     = $data['ts'] ?? null;
        $bizid  = $data['bizid'] ?? null;
        $params = $data['params'] ?? null;
        return true; // Replace with real validation logic.
    }

    /**
     * Perform the actual business work.
     */
    private function doJob($data): bool
    {
        // Insert business logic here.
        return true;
    }
}

Message Creation and Push

In a controller, assemble the payload and push it to a named queue using the ThinkPHP Queue facade.

<?php
declare(strict_types=1);

namespace app\controller;

use app\BaseController;

class Queue extends BaseController
{
    /** Push a new job to the queue. */
    public function push()
    {
        // Handler class that will process the job.
        $jobHandlerClass = \app\job\TinywanJob::class;
        // Queue name – created automatically if it does not exist.
        $queueName = 'tinywanJobQueue';
        // Payload – must be JSON‑serializable.
        $payload = [
            'type'      => 'test',
            'record_id' => 123,
        ];
        // Push the job.
        $result = \think\facade\Queue::push(
            $jobHandlerClass,
            json_encode($payload, JSON_UNESCAPED_UNICODE),
            $queueName
        );
        if ($result !== false) {
            echo date('Y-m-d H:i:s') . ' a new job has been pushed to the queue';
        } else {
            echo 'Failed to push job.';
        }
        return 'push';
    }
}

For larger projects, map a logical message name to a consumer class so producers only need to know the message name, not the concrete handler.

Task Processing

Run a worker process from the project root. The command below starts a long‑running worker that continuously consumes jobs from tinywanJobQueue: php think queue:work --queue tinywanJobQueue In production environments, use a process manager such as Supervisor to keep the worker alive and automatically restart it on failure.

Task Publishing (API Example)

After deploying the controller, invoke the endpoint to push a job: http://127.0.0.1:8786/queue/push The response should indicate whether the job was queued successfully.

Command Reference

Work Mode ( queue:work )

Starts a persistent worker that processes jobs from the specified queue.

php think queue:work --queue tinywanJobQueue

--daemon

: Keep the worker running; omit to process a single job and exit. --queue name : Name of the queue to listen to. --delay seconds : Seconds to wait before retrying a failed job (default 0). --force: Continue processing even when the application is in maintenance mode. --memory megabytes : Memory limit for the worker (e.g., 128). --sleep seconds : Sleep interval when the queue is empty. --tries count : Maximum retry attempts before marking a job as failed.

Listen Mode ( queue:listen )

Spawns a temporary work process for each job, suitable when a persistent worker is not desired.

php think queue:listen --queue tinywanJobQueue

--queue

name : Queue to monitor. --delay seconds : Delay before retrying a failed job. --memory megabytes : Memory limit for each spawned process. --sleep seconds : Interval to wait when the queue is empty. --tries count : Retry limit before moving the job to the failure queue. --timeout seconds : Maximum execution time for a spawned work process.

Both modes can be combined with Supervisor scripts to ensure high availability and automatic restarts.