PHP code example of yangusik / laravel-balanced-queue

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

/* Start to develop here. Best regards https://php-download.com/ */

    

yangusik / laravel-balanced-queue example snippets

'connections' => [
    // ... your existing connections

    'balanced' => [
        'driver' => 'balanced',
        'connection' => 'default', // Redis connection from database.php
        'queue' => 'default',
        'retry_after' => 90,
    ],
],

namespace App\Jobs;

use Illuminate\Contracts\Queue\ShouldQueue;
use YanGusik\BalancedQueue\Jobs\BalancedDispatchable;

class GenerateAIImage implements ShouldQueue
{
    use BalancedDispatchable;  // Instead of standard Dispatchable

    public function __construct(
        public int $userId,
        public string $prompt
    ) {}

    public function handle(): void
    {
        // Your AI generation logic here
    }
}

// The job will automatically use $userId as partition key
GenerateAIImage::dispatch($userId, $prompt)
    ->onConnection('balanced')
    ->onQueue('ai-generation');

// Or explicitly set partition
GenerateAIImage::dispatch($userId, $prompt)
    ->onPartition($userId)
    ->onConnection('balanced');

'environments' => [
    'local' => [
        // Your other supervisors...

        'supervisor-balanced' => [
            'connection' => 'balanced',      // Must match connection name in queue.php
            'queue' => ['default'],          // Queue names to process
            'maxProcesses' => 4,             // Number of workers
            'tries' => 1,
            'timeout' => 300,
        ],
    ],

    'production' => [
        'supervisor-balanced' => [
            'connection' => 'balanced',
            'queue' => ['default', 'ai-generation'],
            'maxProcesses' => 10,
            'tries' => 1,
            'timeout' => 300,
            'balance' => 'auto',             // Horizon's auto-scaling
        ],
    ],
],

// config/balanced-queue.php
'horizon' => [
    'enabled' => 'auto',  // 'auto', true, or false
],

// config/balanced-queue.php
'strategy' => env('BALANCED_QUEUE_STRATEGY', 'round-robin'),

// config/balanced-queue.php
'limiter' => env('BALANCED_QUEUE_LIMITER', 'simple'),

'limiters' => [
    'simple' => [
        'max_concurrent' => 2, // Max 2 jobs per user at once
    ],
],

class MyJob implements ShouldQueue
{
    use BalancedDispatchable;

    public function __construct(
        public int $userId  // Automatically used as partition key
    ) {}
}

// Set partition when dispatching
MyJob::dispatch($data)
    ->onPartition("user:{$userId}")
    ->onConnection('balanced');

// Or set in job constructor
MyJob::dispatch($data)
    ->onPartition($companyId)
    ->onConnection('balanced');

class ProcessOrder implements ShouldQueue
{
    use BalancedDispatchable;

    public function __construct(public Order $order) {}

    public function getPartitionKey(): string
    {
        // Partition by merchant instead of user
        return "merchant:{$this->order->merchant_id}";
    }
}

// config/balanced-queue.php
'partition_resolver' => function ($job) {
    return $job->tenant_id ?? $job->user_id ?? 'default';
},

use YanGusik\BalancedQueue\Support\Metrics;

$metrics = new Metrics();

// Get queue summary
$summary = $metrics->getSummary('default');
// Returns: [
//     'partitions' => 5,
//     'total_queued' => 100,
//     'total_active' => 10,
//     'partitions_stats' => [...]
// ]

// Get per-partition stats
$stats = $metrics->getQueueStats('default');
// Returns: [
//     'user:123' => ['queued' => 10, 'active' => 2, 'metrics' => [...]],
//     'user:456' => ['queued' => 5, 'active' => 1, 'metrics' => [...]],
// ]

// Clear queue programmatically
$metrics->clearQueue('default');

use YanGusik\BalancedQueue\Contracts\PartitionStrategy;
use Illuminate\Contracts\Redis\Connection;

class PriorityStrategy implements PartitionStrategy
{
    public function selectPartition(Connection $redis, string $queue, string $partitionsKey): ?string
    {
        // Get all partitions
        $partitions = $redis->smembers($partitionsKey);

        // Your priority logic here
        // e.g., check user subscription level, queue size, etc.

        return $selectedPartition;
    }

    public function getName(): string
    {
        return 'priority';
    }
}

// config/balanced-queue.php
'strategies' => [
    'priority' => [
        'class' => App\Queue\PriorityStrategy::class,
    ],
],

'strategy' => 'priority',

use YanGusik\BalancedQueue\Contracts\ConcurrencyLimiter;

class PlanBasedLimiter implements ConcurrencyLimiter
{
    public function canProcess($redis, $queue, $partition): bool
    {
        $userId = str_replace('user:', '', $partition);
        $user = User::find($userId);
        $limit = $user->subscription->concurrent_limit ?? 1;

        return $this->getActiveCount($redis, $queue, $partition) < $limit;
    }

    // ... implement other interface methods
}

   ->onConnection('balanced')  // Must match 'balanced' in queue.php
   

'prometheus' => [
    'enabled' => true,
    'middleware' => 'ip_whitelist', // or 'auth.basic', or null
    'ip_whitelist' => [
        '127.0.0.1',
        '10.0.0.0/8',      // Private networks
        '172.16.0.0/12',
        '192.168.0.0/16',
    ],
],

bash
php artisan vendor:publish --tag=balanced-queue-config

bash
   php artisan queue:work balanced
   

bash
php artisan balanced-queue:table --watch

bash
php artisan balanced-queue:table

GET /balanced-queue/metrics/json

Response:
{
  "timestamp": "2024-01-15T10:30:00+00:00",
  "queues": [
    {
      "queue": "default",
      "pending": 84,
      "active": 4,
      "processed": 1250,
      "partition_count": 3,
      "partitions": [
        {"partition": "user:123", "pending": 50, "active": 2, "processed": 800},
        {"partition": "user:456", "pending": 20, "active": 1, "processed": 300},
        {"partition": "user:789", "pending": 14, "active": 1, "processed": 150}
      ]
    }
  ]
}