Asynchronous Tasks (Queue)

Introduction

Asynchronous tasks allow you to defer the execution of heavy processes (sending emails, generating reports, processing images) by placing them in a queue. BowPHP supports several queue backends: Sync, Database, Redis, Beanstalkd, RabbitMQ, Amazon SQS, and Kafka.

Creating a Task

Use the add:task command to generate a new task:

php bow add:task SendWelcomeEmail

This command creates the file app/Tasks/SendWelcomeEmail.php:

app/Tasks/SendWelcomeEmail.php

<?php

namespace App\Tasks;

use Bow\Queue\QueueTask;

class SendWelcomeEmail extends QueueTask
{
  /**
   * The task data
   */
  private string $email;
  private string $name;

  /**
   * Constructor
   */
  public function __construct(string $email, string $name)
  {
    $this->email = $email;
    $this->name = $name;
  }

  /**
   * Executes the task
   */
  public function process(): void
  {
    Mail::send("emails.welcome", ["name" => $this->name], function (Envelop $envelop) {
      $envelop->to($this->email)
              ->subject("Bienvenue, {$this->name} !");
    });
  }
}

Configuration

The queue configuration is located in config/queue.php:

config/queue.php

<?php

return [
  // Default driver
  "default" => app_env("QUEUE_DRIVER", "sync"),

  "connections" => [
    // Synchronous execution (for development)
    "sync" => [
      "queue" => "default",
    ],

    // Database
    "database" => [
      "queue" => "default",
      "table" => "queues",
    ],

    // Redis
    "redis" => [
      "queue" => "default",
      "block_timeout" => 5,
    ],

    // Beanstalkd
    "beanstalkd" => [
      "hostname" => "127.0.0.1",
      "port" => 11300,
      "timeout" => 10,
      "queue" => "default",
    ],

    // RabbitMQ
    "rabbitmq" => [
      "queue" => "default",
      "host" => app_env("RABBITMQ_HOST", "127.0.0.1"),
      "port" => app_env("RABBITMQ_PORT", 5672),
      "user" => app_env("RABBITMQ_USER", "guest"),
      "password" => app_env("RABBITMQ_PASSWORD", "guest"),
      "vhost" => app_env("RABBITMQ_VHOST", "/"),
    ],

    // Amazon SQS
    "sqs" => [
      "queue" => "default",
      "url" => app_env("SQS_URL"),
      "region" => app_env("AWS_REGION"),
      "version" => "latest",
      "credentials" => [
        "key" => app_env("AWS_KEY"),
        "secret" => app_env("AWS_SECRET"),
      ],
    ],

    // Apache Kafka
    "kafka" => [
      "host" => "localhost",
      "port" => 9092,
      "topic" => "default",
      "group_id" => "bow_queue_group",
      "auto_offset_reset" => "earliest",
      "enable_auto_commit" => "true",
    ],
  ],
];

Migration for the database driver

If you use the database driver, create the queue table:

php bow add:migration create_queues_table

migrations/Version_CreateQueuesTable.php

<?php

use Bow\Database\Migration\Table;
use Bow\Database\Migration\Migration;

class Version_CreateQueuesTable extends Migration
{
  public function up(): void
  {
    $this->create("queues", function (Table $table) {
      $table->addUuidPrimary("id");
      $table->addString("queue");
      $table->addLongtext("payload");
      $table->addInteger("attempts", ["default" => 0]);
      $table->addInteger("delay", ["default" => 0]);
      $table->addEnum("status", [
        "size"    => ["pending", "processing", "completed", "failed"],
        "default" => "pending",
      ]);
      $table->addTimestamp("available_at");
      $table->addTimestamps();

      $table->addIndex("queue");
      $table->addIndex("status");
    });
  }

  public function rollback(): void
  {
    $this->dropIfExists("queues");
  }
}

Dispatching a Task

With the helper function

use App\Tasks\SendWelcomeEmail;

// Dispatch immediately
queue(new SendWelcomeEmail("john@example.com", "John"));

From a controller

app/Controllers/UserController.php

<?php

namespace App\Controllers;

use App\Tasks\SendWelcomeEmail;
use App\Models\User;
use Bow\Http\Request;

class UserController extends Controller
{
  public function store(Request $request)
  {
    $user = User::create([
      "name" => $request->get("name"),
      "email" => $request->get("email"),
      "password" => app_hash($request->get("password")),
    ]);

    // Send the email in the background
    queue(new SendWelcomeEmail($user->email, $user->name));

    return response_json([
      "message" => "Utilisateur créé avec succès",
      "user" => $user
    ], 201);
  }
}

Task Properties

Queue

By default, tasks are sent to the default queue. You can specify a different queue:

class ProcessOrderTask extends QueueTask
{
  protected string $queue = "orders";
  
  // ...
}

Execution delay

Defer the execution of a task by a number of seconds:

class SendReminderTask extends QueueTask
{
  protected int $delay = 3600; // 1 hour
  
  // ...
}

Or dynamically:

$task = new SendReminderTask($userId);
$task->setDelay(1800); // 30 minutes
queue($task);

Attempts & retry

Configure the number of attempts and the delay between attempts:

class ProcessPaymentTask extends QueueTask
{
  // Maximum number of attempts
  protected int $attempts = 3;
  
  // Delay between attempts (in seconds)
  protected int $retry = 60;
  
  public function process(): void
  {
    // Process the payment
  }
}

Priority

Tasks with a higher priority are processed first:

class UrgentNotificationTask extends QueueTask
{
  protected int $priority = 10; // High priority (default: 1)
  
  // ...
}

Error Handling

The onException method

Implement onException() to handle errors:

use Throwable;

class ProcessDataTask extends QueueTask
{
  public function process(): void
  {
    // Processing that may fail
  }

  public function onException(Throwable $e): void
  {
    // Log the error
    logger()->error("Échec de ProcessDataTask: " . $e->getMessage());
    
    // Notify the administrator
    Mail::send("emails.task-error", ["error" => $e], function (Envelop $envelop) {
      $envelop->to("admin@example.com")->subject("Erreur de tâche");
    });
  }
}

Deleting a failed task

To prevent further retries:

public function process(): void
{
  try {
    // Processing
  } catch (IrrecoverableException $e) {
    // Do not retry this task
    $this->deleteTask();
    throw $e;
  }
}

Running the Worker

Basic command

# Process the default queue on the default connection
php bow run:worker

# Process a specific queue
php bow run:worker --queue=orders

# Specify a connection (positional argument, not an option)
php bow run:worker redis

# Combine connection + options
php bow run:worker beanstalkd --queue=high --tries=5 --sleep=10

Available options:

Option	Default	Description
--queue	default	Name of the queue to consume
--tries	3	Maximum number of attempts per job
--memory	126	Memory limit in MB
--sleep	3	Seconds to sleep when the queue is empty
--timout	3	Timeout per job (historical spelling of the flag)

In production with Supervisor

Create a Supervisor configuration file:

/etc/supervisor/conf.d/bow-worker.conf

[program:bow-worker]
process_name=%(program_name)s_%(process_num)02d
command=php /var/www/html/bow run:worker --queue=default
autostart=true
autorestart=true
stopasgroup=true
killasgroup=true
user=www-data
numprocs=2
redirect_stderr=true
stdout_logfile=/var/log/bow-worker.log
stopwaitsecs=3600

Start the worker:

sudo supervisorctl reread
sudo supervisorctl update
sudo supervisorctl start bow-worker:*

Complete Example

Report generation task

app/Tasks/GenerateReportTask.php

<?php

namespace App\Tasks;

use Bow\Queue\QueueTask;
use App\Models\Report;
use App\Services\ReportGenerator;
use Throwable;

class GenerateReportTask extends QueueTask
{
  protected string $queue = "reports";
  protected int $attempts = 3;
  protected int $retry = 120;

  private int $reportId;
  private string $format;

  public function __construct(int $reportId, string $format = "pdf")
  {
    $this->reportId = $reportId;
    $this->format = $format;
  }

  public function process(): void
  {
    $report = Report::retrieveOrFail($this->reportId);
    
    $generator = new ReportGenerator();
    $filePath = $generator->generate($report, $this->format);

    $report->update([
      "status" => "completed",
      "file_path" => $filePath,
      "generated_at" => date("Y-m-d H:i:s"),
    ]);

    // Notify the user
    queue(new SendReportReadyNotification(
      $report->user_id,
      $filePath
    ));
  }

  public function onException(Throwable $e): void
  {
    $report = Report::retrieve($this->reportId);
    
    if ($report) {
      $report->update([
        "status" => "failed",
        "error_message" => $e->getMessage(),
      ]);
    }

    logger()->error("Échec génération rapport #{$this->reportId}: " . $e->getMessage());
  }
}

Dispatching from a controller

app/Controllers/ReportController.php

<?php

namespace App\Controllers;

use App\Tasks\GenerateReportTask;
use App\Models\Report;
use Bow\Http\Request;

class ReportController extends Controller
{
  public function generate(Request $request)
  {
    $report = Report::create([
      "user_id" => auth()->id(),
      "type" => $request->get("type"),
      "parameters" => $request->get("parameters"),
      "status" => "pending",
    ]);

    // Dispatch the task
    queue(new GenerateReportTask($report->id, $request->get("format", "pdf")));

    return response_json([
      "message" => "Génération du rapport en cours",
      "report_id" => $report->id,
    ], 202);
  }
}

Integration with the Scheduler

Combine tasks with the scheduler for recurring processing:

app/Kernel.php

public function schedules(Scheduler $schedule): void
{
  // Clean up old sessions every hour
  $schedule->task(App\Tasks\CleanupSessionsTask::class)
    ->hourly()
    ->description("Nettoyer les sessions expirées");

  // Generate daily reports
  $schedule->task(App\Tasks\DailyReportTask::class)
    ->dailyAt("06:00")
    ->description("Générer les rapports quotidiens");
}

Best Practices

Recommendations

• Idempotent tasks: Design your tasks so they can be executed multiple times without side effects.
• Minimal data: Pass only IDs to the constructor, and retrieve the full data in process().
• Error handling: Always implement onException() to log and notify failures.
• Timeouts: Configure appropriate timeouts in Supervisor to avoid stuck tasks.
• Monitoring: Monitor queue size and processing time in production.
• Separate queues: Use distinct queues for critical tasks (payments) and non-critical ones (emails).

Caution

• Tasks are serialized: do not include non-serializable objects (DB connections, closures).
• In sync mode, tasks run synchronously (useful for development).
• Make sure the worker has the same dependencies and configurations as the application.

Is something missing?

If you run into problems with the documentation or have suggestions to improve the documentation or the project in general, please open an issue for us, or send a tweet mentioning the Twitter account @bowframework or directly on github.