Optimizing Laravel Queue Performance with Redis Streams and Advanced Docker Orchestration

Leveraging Redis Streams for High-Throughput Laravel Queues

Traditional Laravel queue drivers, particularly the database and Redis (using lists), can become bottlenecks under heavy load. Redis Streams offer a more robust, persistent, and performant alternative for message queuing, providing features like consumer groups, acknowledgments, and message persistence. This section details the architectural shift required to integrate Redis Streams into your Laravel application.

We’ll be using the predis/predis client, which has excellent support for Redis Streams. First, ensure you have it installed:

composer require predis/predis

Next, configure Laravel to use the Redis Streams driver. This involves modifying your config/queue.php file. We’ll create a new queue connection specifically for Redis Streams.

<?php

return [

    // ... other configurations

    'connections' => [

        // ... other connections

        'redis_streams' => [
            'driver' => 'redis',
            'connection' => 'default', // Assumes a 'default' Redis connection is configured in config/database.php
            'queue' => env('REDIS_STREAMS_QUEUE', 'my_app_streams'), // The name of the stream
            'redis' => [
                'scheme' => env('REDIS_SCHEME', 'tcp'),
                'host' => env('REDIS_HOST', '127.0.0.1'),
                'port' => env('REDIS_PORT', 6379),
                'database' => env('REDIS_STREAMS_DB', 0),
            ],
            'consumer_group' => env('REDIS_STREAMS_CONSUMER_GROUP', 'my_app_consumers'),
            'consumer_name' => env('REDIS_STREAMS_CONSUMER_NAME', 'worker_'.gethostname().'_'.uniqid()),
            'batch_size' => (int) env('REDIS_STREAMS_BATCH_SIZE', 100),
            'retry_delay' => (int) env('REDIS_STREAMS_RETRY_DELAY', 60), // Seconds
        ],

    ],

    // ... other configurations

];

You’ll need to set the corresponding environment variables in your .env file:

REDIS_STREAMS_QUEUE=my_app_streams
REDIS_STREAMS_CONSUMER_GROUP=my_app_consumers
REDIS_STREAMS_CONSUMER_NAME=worker_<your_hostname>_<unique_id> # This will be dynamically set by the worker
REDIS_STREAMS_BATCH_SIZE=100
REDIS_STREAMS_RETRY_DELAY=60

When dispatching jobs, you’ll specify this new connection:

use App\Jobs\ProcessHeavyTask;

ProcessHeavyTask::dispatch()->onConnection('redis_streams');

The worker process needs to be aware of the consumer group and consumer name. The `php artisan queue:work` command supports this natively with the Redis driver. For Redis Streams, it’s crucial to ensure the consumer group is created if it doesn’t exist. The `queue:work` command will handle this automatically on startup if the stream exists. If the stream doesn’t exist, you might need to create it manually or ensure a job is dispatched first.

Advanced Docker Orchestration for Scalable Workers

To effectively manage Redis Streams workers in a production environment, robust Docker orchestration is paramount. We’ll use Docker Compose to define our services: Redis, the Laravel application, and multiple worker instances. The key is to configure the workers for high availability and efficient scaling.

Here’s a sample docker-compose.yml file:

version: '3.8'

services:
  redis:
    image: redis:7-alpine
    container_name: redis_streams_cache
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data
    networks:
      - app_network

  app:
    build:
      context: .
      dockerfile: Dockerfile
    container_name: laravel_app
    volumes:
      - .:/var/www/html
    ports:
      - "8000:8000"
    depends_on:
      - redis
    networks:
      - app_network
    environment:
      - DB_CONNECTION=mysql
      - DB_HOST=mysql
      - DB_PORT=3306
      - DB_DATABASE=myapp
      - DB_USERNAME=user
      - DB_PASSWORD=password
      - REDIS_HOST=redis
      - REDIS_PORT=6379
      - REDIS_STREAMS_QUEUE=my_app_streams
      - REDIS_STREAMS_CONSUMER_GROUP=my_app_consumers
      # REDIS_STREAMS_CONSUMER_NAME will be set dynamically by the worker

  worker_1:
    build:
      context: .
      dockerfile: Dockerfile
    container_name: laravel_worker_1
    command: >
      php artisan queue:work redis_streams
      --tries=3
      --max-time=300
      --sleep=5
      --memory=256
      --daemon
      --env=.env
      --queue=my_app_streams
      --consumer-group=my_app_consumers
      --consumer-name=worker_1_$(hostname)_$(uuidgen)
    volumes:
      - .:/var/www/html
    depends_on:
      - redis
      - app
    networks:
      - app_network
    environment:
      - DB_CONNECTION=mysql
      - DB_HOST=mysql
      - DB_PORT=3306
      - DB_DATABASE=myapp
      - DB_USERNAME=user
      - DB_PASSWORD=password
      - REDIS_HOST=redis
      - REDIS_PORT=6379
      - REDIS_STREAMS_QUEUE=my_app_streams
      - REDIS_STREAMS_CONSUMER_GROUP=my_app_consumers
      # REDIS_STREAMS_CONSUMER_NAME is overridden in the command

  worker_2:
    build:
      context: .
      dockerfile: Dockerfile
    container_name: laravel_worker_2
    command: >
      php artisan queue:work redis_streams
      --tries=3
      --max-time=300
      --sleep=5
      --memory=256
      --daemon
      --env=.env
      --queue=my_app_streams
      --consumer-group=my_app_consumers
      --consumer-name=worker_2_$(hostname)_$(uuidgen)
    volumes:
      - .:/var/www/html
    depends_on:
      - redis
      - app
    networks:
      - app_network
    environment:
      - DB_CONNECTION=mysql
      - DB_HOST=mysql
      - DB_PORT=3306
      - DB_DATABASE=myapp
      - DB_USERNAME=user
      - DB_PASSWORD=password
      - REDIS_HOST=redis
      - REDIS_PORT=6379
      - REDIS_STREAMS_QUEUE=my_app_streams
      - REDIS_STREAMS_CONSUMER_GROUP=my_app_consumers
      # REDIS_STREAMS_CONSUMER_NAME is overridden in the command

  # Add more worker services as needed for scaling

  mysql: # Example MySQL service
    image: mysql:8.0
    container_name: mysql_db
    ports:
      - "3306:3306"
    volumes:
      - mysql_data:/var/lib/mysql
    environment:
      MYSQL_ROOT_PASSWORD: rootpassword
      MYSQL_DATABASE: myapp
      MYSQL_USER: user
      MYSQL_PASSWORD: password
    networks:
      - app_network

volumes:
  redis_data:
  mysql_data:

networks:
  app_network:
    driver: bridge

In this setup:

• We define a redis service using the official Redis image.
• The app service represents your Laravel application, useful for running artisan commands or serving the web application.
• Multiple worker_X services are defined. Each worker runs the queue:work command with specific configurations.
• Crucially, each worker is assigned a unique --consumer-name. This is essential for Redis Streams consumer groups, allowing Redis to track which messages have been processed by which consumers within a group. We use $(hostname)_$(uuidgen) for dynamic, unique naming.
• The --queue parameter points to the Redis Stream name defined in .env.
• --consumer-group specifies the consumer group name. The first worker to connect to a non-existent group will create it.
• --tries, --max-time, --sleep, and --memory are critical for controlling worker behavior and resource consumption.
• The --daemon flag is often used in production to run workers in the background, though for Docker Compose, it’s often managed by the orchestrator.
• Environment variables are passed to each service, ensuring they can connect to Redis and other dependencies.

To start these services, run:

docker-compose up -d

To scale your workers, you can simply add more worker_X services to your docker-compose.yml or use Docker Swarm/Kubernetes for more advanced orchestration. For instance, to run 10 workers:

# Example for scaling with Docker Compose (less common for production scaling)
# You'd typically use Swarm or Kubernetes for this.
# For demonstration, you could manually copy and rename worker services.

# For Docker Swarm:
# docker stack deploy -c docker-compose.yml myapp_stack

# For Kubernetes:
# You would translate this into Deployments, StatefulSets, and Services.

Monitoring and Managing Redis Streams Consumers

Effective monitoring is key to maintaining a healthy queueing system. Redis Streams provide built-in mechanisms for inspecting consumer activity.

You can use the redis-cli to interact with your streams. First, connect to your Redis instance:

docker exec -it redis_streams_cache redis-cli

Once connected, you can use the following commands:

• List all streams: KEYS *streams* (adjust pattern as needed)
• Get stream information: XINFO STREAM my_app_streams
• List consumer groups for a stream: XINFO GROUPS my_app_streams
• List consumers in a specific group: XINFO CONSUMERS my_app_streams my_app_consumers
• View pending messages for a consumer group: XPENDING my_app_streams my_app_consumers
• Claim pending messages (if a consumer dies): XCLAIM my_app_streams my_app_consumers <consumer-name> <min-idle-time> <message-id> ...

For automated monitoring, consider integrating tools like Prometheus with a Redis exporter, or use specialized queue monitoring dashboards that support Redis Streams.

A common scenario is a worker crashing. When this happens, messages it was processing might remain in the pending list. You can manually claim these messages or implement an automated process. A dedicated “claimer” worker or a scheduled task can periodically check the pending list and claim messages that have been idle for too long.

// Example of a scheduled command to claim pending messages
// app/Console/Commands/ClaimStaleJobs.php

namespace App\Console\Commands;

use Illuminate\Console\Command;
use Illuminate\Support\Facades\Redis;
use Illuminate\Support\Str;

class ClaimStaleJobs extends Command
{
    protected $signature = 'queue:claim-stale {stream} {group} {--idle-time=300} {--limit=100}';
    protected $description = 'Claim stale pending messages from a Redis Stream consumer group';

    public function handle()
    {
        $stream = $this->argument('stream');
        $group = $this->argument('group');
        $idleTime = (int) $this->option('idle-time'); // in milliseconds
        $limit = (int) $this->option('limit');

        $redis = Redis::connection('redis_streams'); // Use the correct connection

        $pending = $redis->xpending($stream, $group, '-', '+', $limit);

        if (empty($pending['consumers'])) {
            $this->info("No consumers found for group {$group} on stream {$stream}.");
            return 0;
        }

        $staleMessages = [];
        foreach ($pending['consumers'] as $consumerInfo) {
            if ($consumerInfo['idle'] > $idleTime) {
                $messages = $redis->xreadgroup('GROUP', $group, 'worker_claimer_'.Str::random(8), ['COUNT' => $limit, 'BLOCK' => 0], $stream);
                if ($messages) {
                    foreach ($messages[$stream] as $message) {
                        $messageId = $message[0];
                        $messageData = $message[1];
                        // Check if this message is truly stale based on its own delivery time if available,
                        // or simply claim it if the consumer is stale.
                        // For simplicity here, we assume if the consumer is stale, its messages are too.
                        $staleMessages[] = $messageId;
                        $this->line("Claiming stale message ID: {$messageId} from consumer {$consumerInfo['name']}");
                    }
                }
            }
        }

        if (!empty($staleMessages)) {
            // The XCLAIM command requires a consumer name that is *currently* processing the message.
            // This is a simplification. A more robust approach would involve checking message delivery times.
            // For this example, we'll use a placeholder consumer name.
            // A better approach might be to use XAUTOCLAIM.

            // Using XAUTOCLAIM is generally preferred for automated claiming.
            // XAUTOCLAIM stream group consumer-name min-idle-time start-id [ЕНИЕ] [COUNT count] [JUSTID]
            $claimed = $redis->xautoclaim(
                $stream,
                $group,
                'auto_claimer_'.Str::random(8), // A temporary consumer name for the claim operation
                $idleTime,
                '0-0', // Start ID, '0-0' means from the beginning
                'COUNT', $limit
            );

            if (!empty($claimed[0])) { // claimed[0] contains the new messages
                $this->info("Successfully claimed " . count($claimed[0]) . " messages.");
                foreach ($claimed[0] as $claimedMessage) {
                    $this->line("Claimed message ID: " . $claimedMessage[0]);
                    // You might want to re-dispatch these jobs or process them directly.
                    // For re-dispatching, you'd need to deserialize the job data.
                }
            } else {
                $this->info("No messages claimed.");
            }

        } else {
            $this->info("No stale messages found.");
        }

        return 0;
    }
}

Register this command in app/Console/Kernel.php and schedule it in your Kernel.php, for example, to run every 5 minutes:

// app/Console/Kernel.php
protected function schedule(Schedule $schedule)
{
    $schedule->command('queue:claim-stale my_app_streams my_app_consumers --idle-time=300000 --limit=50')->everyFiveMinutes();
}

This comprehensive approach, combining Redis Streams for message handling and advanced Docker orchestration for worker management, provides a scalable, resilient, and high-performance queuing solution for demanding Laravel applications.