1136 lines
27 KiB
Markdown
1136 lines
27 KiB
Markdown
# Queue System
|
|
|
|
Comprehensive documentation of the asynchronous job processing system in the Custom PHP Framework.
|
|
|
|
## Overview
|
|
|
|
The Queue System provides robust, scalable background job processing with support for multiple queue drivers, priority-based execution, retry mechanisms, dead letter queues, and comprehensive monitoring.
|
|
|
|
**Core Features**:
|
|
- **Multi-Driver Support**: File-based, Redis, and Database-backed queues
|
|
- **Priority Queues**: Critical, high, normal, low priority levels
|
|
- **Delayed Jobs**: Schedule jobs for future execution
|
|
- **Retry Strategies**: Exponential backoff with configurable attempts
|
|
- **Dead Letter Queue**: Automatic failed job management
|
|
- **Job Persistence**: Full job lifecycle tracking and history
|
|
- **Progress Tracking**: Real-time job progress monitoring
|
|
- **Distributed Locking**: Multi-worker coordination
|
|
- **Metrics & Monitoring**: Performance statistics and health checks
|
|
|
|
## Queue Interface
|
|
|
|
The framework provides a unified `Queue` interface that all queue drivers implement:
|
|
|
|
```php
|
|
interface Queue
|
|
{
|
|
/**
|
|
* Push a job to the queue
|
|
*/
|
|
public function push(JobPayload $payload): void;
|
|
|
|
/**
|
|
* Pop the next available job from the queue
|
|
*/
|
|
public function pop(): ?JobPayload;
|
|
|
|
/**
|
|
* Peek at the next job without removing it
|
|
*/
|
|
public function peek(): ?JobPayload;
|
|
|
|
/**
|
|
* Get the number of jobs in the queue
|
|
*/
|
|
public function size(): int;
|
|
|
|
/**
|
|
* Clear all jobs from the queue
|
|
*/
|
|
public function clear(): int;
|
|
|
|
/**
|
|
* Get queue statistics
|
|
*/
|
|
public function getStats(): array;
|
|
}
|
|
```
|
|
|
|
### JobPayload Value Object
|
|
|
|
All jobs are wrapped in a `JobPayload` value object that provides type safety and configuration:
|
|
|
|
```php
|
|
final readonly class JobPayload
|
|
{
|
|
public function __construct(
|
|
public object $job, // The job instance
|
|
public QueuePriority $priority, // Priority level
|
|
public Duration $delay, // Execution delay
|
|
public ?Duration $timeout = null, // Max execution time
|
|
public ?RetryStrategy $retryStrategy = null, // Retry configuration
|
|
public ?JobMetadata $metadata = null // Additional metadata
|
|
) {}
|
|
}
|
|
```
|
|
|
|
**Factory Methods**:
|
|
```php
|
|
// Immediate execution with high priority
|
|
$payload = JobPayload::immediate($job);
|
|
|
|
// Delayed execution
|
|
$payload = JobPayload::delayed($job, Duration::fromMinutes(5));
|
|
|
|
// Critical priority with timeout
|
|
$payload = JobPayload::critical($job);
|
|
|
|
// Background job with retries
|
|
$payload = JobPayload::background($job);
|
|
|
|
// Custom configuration
|
|
$payload = JobPayload::create(
|
|
job: $job,
|
|
priority: QueuePriority::high(),
|
|
delay: Duration::fromSeconds(30),
|
|
timeout: Duration::fromMinutes(5),
|
|
retryStrategy: new ExponentialBackoffStrategy(maxAttempts: 3)
|
|
);
|
|
```
|
|
|
|
## Queue Drivers
|
|
|
|
### File-Based Queue (Default)
|
|
|
|
Simple file-based queue implementation for development and small-scale applications:
|
|
|
|
**Features**:
|
|
- No external dependencies
|
|
- Persistent across restarts
|
|
- Suitable for development and small production loads
|
|
- Priority-based sorting
|
|
|
|
**Usage**:
|
|
```php
|
|
use App\Framework\Queue\FileQueue;
|
|
|
|
$queue = new FileQueue(
|
|
directory: '/var/www/storage/queues',
|
|
queueName: 'default'
|
|
);
|
|
|
|
$queue->push(JobPayload::immediate($job));
|
|
```
|
|
|
|
### Redis Queue (Production)
|
|
|
|
High-performance Redis-backed queue with advanced features:
|
|
|
|
**Features**:
|
|
- High throughput (10,000+ jobs/second)
|
|
- Redis sorted sets for priority queues
|
|
- Delayed job support via timestamp scoring
|
|
- Atomic operations for thread safety
|
|
- Real-time statistics
|
|
|
|
**Architecture**:
|
|
- Priority Queue: Redis sorted set with score = priority + timestamp
|
|
- Delayed Queue: Redis sorted set with score = execution timestamp
|
|
- Statistics: Redis hash with counters
|
|
|
|
**Usage**:
|
|
```php
|
|
use App\Framework\Queue\RedisQueue;
|
|
|
|
$queue = new RedisQueue(
|
|
connection: $redisConnection,
|
|
queueName: 'production',
|
|
serializer: new PhpSerializer()
|
|
);
|
|
|
|
$queue->push(JobPayload::immediate($job));
|
|
|
|
// Statistics
|
|
$stats = $queue->getStats();
|
|
// [
|
|
// 'total_size' => 1523,
|
|
// 'priority_queue_size' => 1200,
|
|
// 'delayed_queue_size' => 323,
|
|
// 'priority_breakdown' => ['high' => 150, 'normal' => 1050, 'low' => 323]
|
|
// ]
|
|
```
|
|
|
|
**Score Calculation**:
|
|
```php
|
|
// Lower score = higher priority in Redis
|
|
$priorityScore = 1000 - $priority->value; // Invert priority
|
|
$timeScore = $currentTime / 1000000; // Microsecond precision for FIFO
|
|
$finalScore = $priorityScore + $timeScore;
|
|
```
|
|
|
|
**Delayed Job Processing**:
|
|
```php
|
|
// Automatic processing of ready delayed jobs
|
|
private function processDelayedJobs(): void
|
|
{
|
|
$currentTime = time();
|
|
|
|
// Get all delayed jobs where score <= current time
|
|
$readyJobs = $this->redis->zRangeByScore($this->delayedKey, 0, $currentTime);
|
|
|
|
foreach ($readyJobs as $serializedPayload) {
|
|
// Remove from delayed queue
|
|
$this->redis->zRem($this->delayedKey, $serializedPayload);
|
|
|
|
// Add to priority queue
|
|
$this->redis->zAdd($this->priorityKey, $score, $serializedPayload);
|
|
}
|
|
}
|
|
```
|
|
|
|
### Persistent Queue Decorator
|
|
|
|
Wraps any queue implementation with job persistence and tracking:
|
|
|
|
```php
|
|
final readonly class PersistentQueue implements Queue
|
|
{
|
|
public function __construct(
|
|
private Queue $baseQueue,
|
|
private JobPersistenceLayer $persistence,
|
|
private QueueType $queueType
|
|
) {}
|
|
|
|
public function push(JobPayload $payload): void
|
|
{
|
|
// Generate job ID
|
|
$jobId = JobId::generate();
|
|
|
|
// Store in persistence layer
|
|
$this->persistence->storeJob(
|
|
jobId: $jobId,
|
|
queueType: $this->queueType,
|
|
jobData: $payload->job,
|
|
maxAttempts: $payload->retryStrategy->maxAttempts
|
|
);
|
|
|
|
// Push to base queue with job ID
|
|
$this->baseQueue->push($payload->withJobId($jobId));
|
|
}
|
|
|
|
public function markJobCompleted(JobId $jobId, array $result = []): void
|
|
{
|
|
$this->persistence->markAsCompleted($jobId, $result);
|
|
}
|
|
|
|
public function markJobFailed(JobId $jobId, string $errorMessage): void
|
|
{
|
|
$jobState = $this->persistence->markAsFailed($jobId, $errorMessage);
|
|
|
|
// Automatic retry if applicable
|
|
if ($jobState->canRetry()) {
|
|
$this->baseQueue->push($retryPayload);
|
|
$this->persistence->markForRetry($jobId, $errorMessage);
|
|
}
|
|
}
|
|
}
|
|
```
|
|
|
|
## Background Jobs
|
|
|
|
### Creating a Job
|
|
|
|
Jobs are simple PHP classes with a `handle()` method:
|
|
|
|
```php
|
|
final readonly class SendWelcomeEmailJob
|
|
{
|
|
public function __construct(
|
|
private UserId $userId,
|
|
private Email $email,
|
|
private UserName $userName
|
|
) {}
|
|
|
|
public function handle(): array
|
|
{
|
|
// Business logic
|
|
$this->emailService->send(
|
|
to: $this->email,
|
|
template: 'welcome',
|
|
data: ['name' => $this->userName->value]
|
|
);
|
|
|
|
// Return result for logging
|
|
return [
|
|
'user_id' => $this->userId->toString(),
|
|
'email' => $this->email->getMasked(),
|
|
'sent_at' => time()
|
|
];
|
|
}
|
|
|
|
public function getType(): string
|
|
{
|
|
return 'email.welcome';
|
|
}
|
|
}
|
|
```
|
|
|
|
### Dispatching Jobs
|
|
|
|
```php
|
|
use App\Framework\Queue\Queue;
|
|
use App\Framework\Queue\ValueObjects\JobPayload;
|
|
|
|
// Immediate execution
|
|
$job = new SendWelcomeEmailJob($userId, $email, $userName);
|
|
$queue->push(JobPayload::immediate($job));
|
|
|
|
// Delayed execution (5 minutes)
|
|
$queue->push(JobPayload::delayed($job, Duration::fromMinutes(5)));
|
|
|
|
// Background job with retries
|
|
$queue->push(JobPayload::background($job));
|
|
|
|
// Custom configuration
|
|
$queue->push(JobPayload::create(
|
|
job: $job,
|
|
priority: QueuePriority::high(),
|
|
timeout: Duration::fromMinutes(5),
|
|
retryStrategy: new ExponentialBackoffStrategy(maxAttempts: 3)
|
|
));
|
|
```
|
|
|
|
### Processing Jobs
|
|
|
|
```php
|
|
// Worker loop
|
|
while (true) {
|
|
$payload = $queue->pop();
|
|
|
|
if ($payload === null) {
|
|
sleep(1); // Wait for new jobs
|
|
continue;
|
|
}
|
|
|
|
try {
|
|
// Execute job
|
|
$result = $payload->job->handle();
|
|
|
|
// Mark as completed
|
|
if ($queue instanceof PersistentQueue) {
|
|
$queue->markJobCompleted($payload->getJobId(), $result);
|
|
}
|
|
} catch (\Throwable $e) {
|
|
// Mark as failed
|
|
if ($queue instanceof PersistentQueue) {
|
|
$queue->markJobFailed(
|
|
$payload->getJobId(),
|
|
$e->getMessage(),
|
|
$e
|
|
);
|
|
}
|
|
}
|
|
}
|
|
```
|
|
|
|
## Priority System
|
|
|
|
### QueuePriority Value Object
|
|
|
|
```php
|
|
enum QueuePriority: int
|
|
{
|
|
case CRITICAL = 1000; // System-critical operations
|
|
case HIGH = 750; // User-facing operations
|
|
case NORMAL = 500; // Standard background tasks
|
|
case LOW = 250; // Non-urgent maintenance
|
|
|
|
public static function critical(): self { return self::CRITICAL; }
|
|
public static function high(): self { return self::HIGH; }
|
|
public static function normal(): self { return self::NORMAL; }
|
|
public static function low(): self { return self::LOW; }
|
|
|
|
public function toString(): string
|
|
{
|
|
return match ($this) {
|
|
self::CRITICAL => 'critical',
|
|
self::HIGH => 'high',
|
|
self::NORMAL => 'normal',
|
|
self::LOW => 'low',
|
|
};
|
|
}
|
|
}
|
|
```
|
|
|
|
### Priority Usage Examples
|
|
|
|
```php
|
|
// Critical: Payment processing, real-time notifications
|
|
$queue->push(JobPayload::create(
|
|
job: new ProcessPaymentJob($payment),
|
|
priority: QueuePriority::critical(),
|
|
timeout: Duration::fromSeconds(30)
|
|
));
|
|
|
|
// High: User-triggered emails, report generation
|
|
$queue->push(JobPayload::create(
|
|
job: new GenerateReportJob($userId),
|
|
priority: QueuePriority::high()
|
|
));
|
|
|
|
// Normal: Scheduled maintenance, cache warming
|
|
$queue->push(JobPayload::create(
|
|
job: new WarmCacheJob(),
|
|
priority: QueuePriority::normal()
|
|
));
|
|
|
|
// Low: Cleanup tasks, analytics aggregation
|
|
$queue->push(JobPayload::create(
|
|
job: new CleanupOldLogsJob(),
|
|
priority: QueuePriority::low()
|
|
));
|
|
```
|
|
|
|
## Retry Mechanisms
|
|
|
|
### RetryStrategy Interface
|
|
|
|
```php
|
|
interface RetryStrategy
|
|
{
|
|
public function shouldRetry(int $attempts): bool;
|
|
public function getDelay(int $attempts): Duration;
|
|
public function getMaxAttempts(): int;
|
|
}
|
|
```
|
|
|
|
### Exponential Backoff Strategy
|
|
|
|
```php
|
|
final readonly class ExponentialBackoffStrategy implements RetryStrategy
|
|
{
|
|
public function __construct(
|
|
private int $maxAttempts = 3,
|
|
private int $baseDelaySeconds = 60,
|
|
private int $maxDelaySeconds = 3600
|
|
) {}
|
|
|
|
public function shouldRetry(int $attempts): bool
|
|
{
|
|
return $attempts < $this->maxAttempts;
|
|
}
|
|
|
|
public function getDelay(int $attempts): Duration
|
|
{
|
|
// 2^attempts * baseDelay, capped at maxDelay
|
|
$delaySeconds = min(
|
|
pow(2, $attempts) * $this->baseDelaySeconds,
|
|
$this->maxDelaySeconds
|
|
);
|
|
|
|
return Duration::fromSeconds($delaySeconds);
|
|
}
|
|
|
|
public function getMaxAttempts(): int
|
|
{
|
|
return $this->maxAttempts;
|
|
}
|
|
}
|
|
```
|
|
|
|
### Custom Retry Strategy
|
|
|
|
```php
|
|
final readonly class CustomRetryStrategy implements RetryStrategy
|
|
{
|
|
public function __construct(
|
|
private array $retryDelays = [60, 300, 1800] // 1min, 5min, 30min
|
|
) {}
|
|
|
|
public function shouldRetry(int $attempts): bool
|
|
{
|
|
return $attempts < count($this->retryDelays);
|
|
}
|
|
|
|
public function getDelay(int $attempts): Duration
|
|
{
|
|
$index = min($attempts, count($this->retryDelays) - 1);
|
|
return Duration::fromSeconds($this->retryDelays[$index]);
|
|
}
|
|
|
|
public function getMaxAttempts(): int
|
|
{
|
|
return count($this->retryDelays);
|
|
}
|
|
}
|
|
```
|
|
|
|
### Usage with Jobs
|
|
|
|
```php
|
|
// Standard exponential backoff
|
|
$payload = JobPayload::create(
|
|
job: $job,
|
|
retryStrategy: new ExponentialBackoffStrategy(
|
|
maxAttempts: 5,
|
|
baseDelaySeconds: 60, // Start with 1 minute
|
|
maxDelaySeconds: 3600 // Cap at 1 hour
|
|
)
|
|
);
|
|
|
|
// Custom retry schedule
|
|
$payload = JobPayload::create(
|
|
job: $job,
|
|
retryStrategy: new CustomRetryStrategy([60, 300, 900, 3600])
|
|
);
|
|
|
|
// No retries
|
|
$payload = JobPayload::create(
|
|
job: $job,
|
|
retryStrategy: null // Job fails permanently on first error
|
|
);
|
|
```
|
|
|
|
## Failed Jobs & Dead Letter Queue
|
|
|
|
### JobStatus Lifecycle
|
|
|
|
```php
|
|
enum JobStatus: string
|
|
{
|
|
case PENDING = 'pending'; // Waiting in queue
|
|
case PROCESSING = 'processing'; // Currently executing
|
|
case COMPLETED = 'completed'; // Successfully completed
|
|
case FAILED = 'failed'; // Failed with error
|
|
case RETRYING = 'retrying'; // Failed, will retry
|
|
case CANCELLED = 'cancelled'; // Manually cancelled
|
|
case EXPIRED = 'expired'; // Timeout exceeded
|
|
|
|
public function isFinal(): bool
|
|
{
|
|
return match ($this) {
|
|
self::COMPLETED, self::CANCELLED, self::EXPIRED => true,
|
|
default => false
|
|
};
|
|
}
|
|
|
|
public function isActive(): bool
|
|
{
|
|
return match ($this) {
|
|
self::PENDING, self::PROCESSING, self::RETRYING => true,
|
|
default => false
|
|
};
|
|
}
|
|
|
|
public function isFailure(): bool
|
|
{
|
|
return match ($this) {
|
|
self::FAILED, self::EXPIRED => true,
|
|
default => false
|
|
};
|
|
}
|
|
}
|
|
```
|
|
|
|
### Dead Letter Queue Manager
|
|
|
|
When jobs exceed max retry attempts, they are automatically moved to a dead letter queue:
|
|
|
|
```php
|
|
final readonly class DeadLetterManager
|
|
{
|
|
public function __construct(
|
|
private DeadLetterQueueInterface $deadLetterQueue,
|
|
private ProductionJobPersistenceLayer $persistenceLayer
|
|
) {}
|
|
|
|
/**
|
|
* Move failed job to dead letter queue
|
|
*/
|
|
public function moveJobToDeadLetterQueue(
|
|
JobIndexEntry $failedJob,
|
|
FailureReason $failureReason
|
|
): void {
|
|
$dlqName = DeadLetterQueueName::forQueue(
|
|
QueueName::fromString($failedJob->queueType)
|
|
);
|
|
|
|
$deadLetterJob = DeadLetterJob::fromFailedJob(
|
|
failedJob: $failedJob,
|
|
deadLetterQueueName: $dlqName,
|
|
failureReason: $failureReason
|
|
);
|
|
|
|
$this->deadLetterQueue->addFailedJob($deadLetterJob);
|
|
$this->persistenceLayer->deleteJob($failedJob->jobId);
|
|
}
|
|
|
|
/**
|
|
* Handle job failure with automatic DLQ routing
|
|
*/
|
|
public function handleJobFailure(
|
|
string $jobId,
|
|
\Throwable $exception,
|
|
int $maxAttempts = 3
|
|
): bool {
|
|
$jobEntry = $this->persistenceLayer->getJobById($jobId);
|
|
|
|
if ($jobEntry->attempts >= $maxAttempts) {
|
|
$this->moveJobToDeadLetterQueue(
|
|
failedJob: $jobEntry,
|
|
failureReason: FailureReason::fromException($exception)
|
|
);
|
|
|
|
return true;
|
|
}
|
|
|
|
return false;
|
|
}
|
|
}
|
|
```
|
|
|
|
### Managing Failed Jobs
|
|
|
|
```php
|
|
use App\Framework\Queue\Services\DeadLetterManager;
|
|
|
|
// Get failed jobs
|
|
$failedJobs = $deadLetterManager->getFailedJobs(
|
|
originalQueue: QueueName::fromString('emails'),
|
|
limit: 100
|
|
);
|
|
|
|
// Retry a specific job
|
|
$deadLetterManager->retryJob($deadLetterJobId);
|
|
|
|
// Retry all jobs in a dead letter queue
|
|
$retriedCount = $deadLetterManager->retryAllJobs($deadLetterQueueName);
|
|
|
|
// Clear dead letter queue
|
|
$deletedCount = $deadLetterManager->clearDeadLetterQueue($deadLetterQueueName);
|
|
|
|
// Get statistics
|
|
$stats = $deadLetterManager->getStatistics();
|
|
// [
|
|
// 'default.dlq' => ['count' => 23, 'oldest' => '2024-01-15 10:30:00'],
|
|
// 'emails.dlq' => ['count' => 5, 'oldest' => '2024-01-16 14:20:00']
|
|
// ]
|
|
```
|
|
|
|
## Queue Monitoring
|
|
|
|
### Queue Statistics
|
|
|
|
```php
|
|
// Get comprehensive statistics
|
|
$stats = $queue->getStats();
|
|
|
|
// Example output:
|
|
[
|
|
'total_size' => 1523,
|
|
'priority_queue_size' => 1200,
|
|
'delayed_queue_size' => 323,
|
|
'priority_breakdown' => [
|
|
'critical' => 15,
|
|
'high' => 150,
|
|
'normal' => 1050,
|
|
'low' => 308
|
|
],
|
|
'stats' => [
|
|
'total_pushed' => 125430,
|
|
'total_popped' => 123907,
|
|
'high_pushed' => 15234,
|
|
'normal_pushed' => 95678,
|
|
'low_pushed' => 14518,
|
|
'last_activity' => 1705410234
|
|
],
|
|
'persistence' => [
|
|
'total_jobs' => 125430,
|
|
'completed' => 120145,
|
|
'failed' => 3762,
|
|
'retrying' => 1523
|
|
]
|
|
]
|
|
```
|
|
|
|
### Job Metrics
|
|
|
|
```php
|
|
use App\Framework\Queue\Services\JobMetricsManager;
|
|
|
|
// Get job performance metrics
|
|
$metrics = $metricsManager->getJobMetrics($jobId);
|
|
|
|
// [
|
|
// 'execution_time_ms' => 1234,
|
|
// 'memory_used_mb' => 45.2,
|
|
// 'peak_memory_mb' => 52.8,
|
|
// 'attempts' => 1,
|
|
// 'last_attempt_at' => '2024-01-16 15:30:00'
|
|
// ]
|
|
|
|
// Get aggregated metrics by job type
|
|
$aggregated = $metricsManager->getAggregatedMetrics('email.welcome');
|
|
|
|
// [
|
|
// 'total_executions' => 1523,
|
|
// 'avg_execution_time_ms' => 856,
|
|
// 'p95_execution_time_ms' => 1450,
|
|
// 'p99_execution_time_ms' => 2340,
|
|
// 'success_rate' => 0.987,
|
|
// 'avg_memory_mb' => 38.4
|
|
// ]
|
|
```
|
|
|
|
### Progress Tracking
|
|
|
|
For long-running jobs, track progress:
|
|
|
|
```php
|
|
final readonly class LargeDataImportJob
|
|
{
|
|
public function __construct(
|
|
private ProgressManager $progressManager,
|
|
private JobId $jobId,
|
|
private array $dataFiles
|
|
) {}
|
|
|
|
public function handle(): array
|
|
{
|
|
$totalSteps = count($this->dataFiles);
|
|
|
|
foreach ($this->dataFiles as $index => $file) {
|
|
// Process file
|
|
$this->processFile($file);
|
|
|
|
// Update progress
|
|
$this->progressManager->updateProgress(
|
|
jobId: $this->jobId,
|
|
currentStep: $index + 1,
|
|
totalSteps: $totalSteps,
|
|
message: "Processing {$file}"
|
|
);
|
|
}
|
|
|
|
return ['files_processed' => $totalSteps];
|
|
}
|
|
}
|
|
|
|
// Query progress
|
|
$progress = $progressManager->getProgress($jobId);
|
|
|
|
// [
|
|
// 'current_step' => 45,
|
|
// 'total_steps' => 100,
|
|
// 'percentage' => 45.0,
|
|
// 'message' => 'Processing data_2024_01_16.csv',
|
|
// 'updated_at' => '2024-01-16 15:32:45'
|
|
// ]
|
|
```
|
|
|
|
### Health Monitoring
|
|
|
|
```php
|
|
use App\Framework\Queue\Services\WorkerHealthCheckService;
|
|
|
|
// Check worker health
|
|
$health = $healthCheckService->checkAllWorkers();
|
|
|
|
// [
|
|
// 'healthy_workers' => 8,
|
|
// 'unhealthy_workers' => 1,
|
|
// 'total_workers' => 9,
|
|
// 'workers' => [
|
|
// ['id' => 'worker-1', 'status' => 'healthy', 'last_heartbeat' => '2024-01-16 15:33:00'],
|
|
// ['id' => 'worker-2', 'status' => 'unhealthy', 'last_heartbeat' => '2024-01-16 15:20:15']
|
|
// ]
|
|
// ]
|
|
|
|
// Register worker heartbeat
|
|
$healthCheckService->registerHeartbeat($workerId);
|
|
|
|
// Cleanup stale workers
|
|
$cleanedCount = $healthCheckService->cleanupStaleWorkers(
|
|
staleThreshold: Duration::fromMinutes(5)
|
|
);
|
|
```
|
|
|
|
## Best Practices
|
|
|
|
### 1. Job Design
|
|
|
|
**Keep Jobs Small and Focused**:
|
|
```php
|
|
// ✅ Good: Single responsibility
|
|
final readonly class SendEmailJob
|
|
{
|
|
public function handle(): array
|
|
{
|
|
$this->emailService->send($this->email, $this->template);
|
|
return ['sent' => true];
|
|
}
|
|
}
|
|
|
|
// ❌ Bad: Multiple responsibilities
|
|
final readonly class SendEmailAndUpdateUserJob
|
|
{
|
|
public function handle(): array
|
|
{
|
|
$this->emailService->send($this->email);
|
|
$this->userRepository->update($this->userId); // Separate concern
|
|
$this->analyticsService->track($this->event); // Separate concern
|
|
}
|
|
}
|
|
```
|
|
|
|
**Make Jobs Idempotent**:
|
|
```php
|
|
final readonly class ProcessPaymentJob
|
|
{
|
|
public function handle(): array
|
|
{
|
|
// Check if already processed (idempotency)
|
|
if ($this->paymentRepository->isProcessed($this->paymentId)) {
|
|
return ['status' => 'already_processed'];
|
|
}
|
|
|
|
// Process payment
|
|
$result = $this->paymentGateway->charge($this->payment);
|
|
|
|
// Mark as processed
|
|
$this->paymentRepository->markAsProcessed($this->paymentId);
|
|
|
|
return $result;
|
|
}
|
|
}
|
|
```
|
|
|
|
**Use Value Objects for Job Data**:
|
|
```php
|
|
// ✅ Good: Type-safe value objects
|
|
final readonly class SendInvoiceJob
|
|
{
|
|
public function __construct(
|
|
private InvoiceId $invoiceId,
|
|
private Email $recipientEmail,
|
|
private Money $amount
|
|
) {}
|
|
}
|
|
|
|
// ❌ Bad: Primitive arrays
|
|
final readonly class SendInvoiceJob
|
|
{
|
|
public function __construct(
|
|
private array $data // What structure? What validation?
|
|
) {}
|
|
}
|
|
```
|
|
|
|
### 2. Priority Assignment
|
|
|
|
**Critical**: System-critical operations that must execute immediately
|
|
- Payment processing
|
|
- Real-time notifications
|
|
- Security alerts
|
|
|
|
**High**: User-facing operations with user waiting
|
|
- Password reset emails
|
|
- Welcome emails
|
|
- Report generation (user-triggered)
|
|
|
|
**Normal**: Standard background tasks
|
|
- Scheduled maintenance
|
|
- Cache warming
|
|
- Analytics processing
|
|
|
|
**Low**: Non-urgent maintenance
|
|
- Log cleanup
|
|
- Old data archival
|
|
- Non-critical aggregations
|
|
|
|
### 3. Retry Strategy
|
|
|
|
**Use Exponential Backoff** for transient failures:
|
|
```php
|
|
// External API calls, network operations
|
|
JobPayload::create(
|
|
job: $apiCallJob,
|
|
retryStrategy: new ExponentialBackoffStrategy(
|
|
maxAttempts: 5,
|
|
baseDelaySeconds: 60,
|
|
maxDelaySeconds: 3600
|
|
)
|
|
);
|
|
```
|
|
|
|
**No Retries** for permanent failures:
|
|
```php
|
|
// Invalid input, business logic violations
|
|
JobPayload::create(
|
|
job: $validationJob,
|
|
retryStrategy: null // Fail immediately
|
|
);
|
|
```
|
|
|
|
**Custom Schedule** for specific requirements:
|
|
```php
|
|
// Time-sensitive operations
|
|
JobPayload::create(
|
|
job: $timeSensitiveJob,
|
|
retryStrategy: new CustomRetryStrategy([30, 60, 120]) // Quick retries
|
|
);
|
|
```
|
|
|
|
### 4. Monitoring
|
|
|
|
**Track Key Metrics**:
|
|
- Queue depth by priority
|
|
- Average job execution time
|
|
- Success/failure rates
|
|
- Dead letter queue size
|
|
- Worker health status
|
|
|
|
**Set Up Alerts**:
|
|
```php
|
|
// Alert on high queue depth
|
|
if ($stats['total_size'] > 10000) {
|
|
$alerting->send('Queue depth exceeded threshold');
|
|
}
|
|
|
|
// Alert on high failure rate
|
|
$failureRate = $stats['persistence']['failed'] / $stats['persistence']['total_jobs'];
|
|
if ($failureRate > 0.05) { // 5% failure rate
|
|
$alerting->send('High job failure rate detected');
|
|
}
|
|
|
|
// Alert on dead letter queue growth
|
|
if ($dlqStats['count'] > 100) {
|
|
$alerting->send('Dead letter queue requires attention');
|
|
}
|
|
```
|
|
|
|
### 5. Resource Management
|
|
|
|
**Implement Timeouts**:
|
|
```php
|
|
JobPayload::create(
|
|
job: $job,
|
|
timeout: Duration::fromMinutes(5) // Kill job after 5 minutes
|
|
);
|
|
```
|
|
|
|
**Manage Memory**:
|
|
```php
|
|
final readonly class LargeDataProcessingJob
|
|
{
|
|
public function handle(): array
|
|
{
|
|
// Process in batches to avoid memory exhaustion
|
|
foreach ($this->getBatches() as $batch) {
|
|
$this->processBatch($batch);
|
|
|
|
// Force garbage collection
|
|
gc_collect_cycles();
|
|
}
|
|
}
|
|
}
|
|
```
|
|
|
|
**Connection Pooling**:
|
|
```php
|
|
// Reuse connections across jobs
|
|
final readonly class DatabaseJob
|
|
{
|
|
public function handle(): array
|
|
{
|
|
// Use connection pool instead of new connections
|
|
$connection = $this->connectionPool->acquire();
|
|
|
|
try {
|
|
// Execute queries
|
|
$result = $connection->query($sql);
|
|
} finally {
|
|
// Always release back to pool
|
|
$this->connectionPool->release($connection);
|
|
}
|
|
}
|
|
}
|
|
```
|
|
|
|
### 6. Error Handling
|
|
|
|
**Graceful Degradation**:
|
|
```php
|
|
final readonly class EmailJob
|
|
{
|
|
public function handle(): array
|
|
{
|
|
try {
|
|
$this->emailService->send($this->email);
|
|
return ['status' => 'sent'];
|
|
} catch (EmailProviderException $e) {
|
|
// Log error but don't fail job
|
|
$this->logger->warning('Email provider unavailable', [
|
|
'error' => $e->getMessage()
|
|
]);
|
|
|
|
// Use fallback provider
|
|
$this->fallbackEmailService->send($this->email);
|
|
return ['status' => 'sent_via_fallback'];
|
|
}
|
|
}
|
|
}
|
|
```
|
|
|
|
**Detailed Error Logging**:
|
|
```php
|
|
try {
|
|
$job->handle();
|
|
} catch (\Throwable $e) {
|
|
$this->logger->error('Job failed', [
|
|
'job_id' => $jobId,
|
|
'job_class' => get_class($job),
|
|
'error' => $e->getMessage(),
|
|
'trace' => $e->getTraceAsString(),
|
|
'attempts' => $attempts,
|
|
'metadata' => $payload->metadata->toArray()
|
|
]);
|
|
|
|
throw $e;
|
|
}
|
|
```
|
|
|
|
### 7. Testing
|
|
|
|
**Unit Test Jobs**:
|
|
```php
|
|
it('sends welcome email', function () {
|
|
$job = new SendWelcomeEmailJob(
|
|
userId: UserId::generate(),
|
|
email: new Email('test@example.com'),
|
|
userName: new UserName('Test User')
|
|
);
|
|
|
|
$result = $job->handle();
|
|
|
|
expect($result['sent_at'])->toBeGreaterThan(0);
|
|
});
|
|
```
|
|
|
|
**Integration Test Queue Operations**:
|
|
```php
|
|
it('processes jobs in priority order', function () {
|
|
$queue = new RedisQueue($connection);
|
|
|
|
// Push jobs with different priorities
|
|
$queue->push(JobPayload::create($lowJob, QueuePriority::low()));
|
|
$queue->push(JobPayload::create($highJob, QueuePriority::high()));
|
|
$queue->push(JobPayload::create($normalJob, QueuePriority::normal()));
|
|
|
|
// Pop and verify order
|
|
$first = $queue->pop();
|
|
expect($first->job)->toBe($highJob);
|
|
|
|
$second = $queue->pop();
|
|
expect($second->job)->toBe($normalJob);
|
|
|
|
$third = $queue->pop();
|
|
expect($third->job)->toBe($lowJob);
|
|
});
|
|
```
|
|
|
|
## Framework Integration
|
|
|
|
### With Event System
|
|
|
|
```php
|
|
// Dispatch jobs in response to events
|
|
#[OnEvent(priority: 100)]
|
|
public function onUserRegistered(UserRegisteredEvent $event): void
|
|
{
|
|
$job = new SendWelcomeEmailJob(
|
|
userId: $event->userId,
|
|
email: $event->email,
|
|
userName: $event->userName
|
|
);
|
|
|
|
$this->queue->push(JobPayload::immediate($job));
|
|
}
|
|
```
|
|
|
|
### With Scheduler
|
|
|
|
See [Scheduler-Queue Pipeline](scheduler-queue-pipeline.md) for complete integration documentation.
|
|
|
|
```php
|
|
// Schedule recurring job dispatch
|
|
$scheduler->schedule(
|
|
'send-daily-digest',
|
|
CronSchedule::fromExpression('0 8 * * *'), // Every day at 8 AM
|
|
fn() => $this->queue->push(JobPayload::immediate(new SendDailyDigestJob()))
|
|
);
|
|
```
|
|
|
|
### With Command Bus
|
|
|
|
```php
|
|
// Dispatch jobs via command handlers
|
|
final readonly class CreateUserHandler
|
|
{
|
|
public function handle(CreateUserCommand $command): User
|
|
{
|
|
$user = $this->userRepository->create($command);
|
|
|
|
// Queue welcome email
|
|
$this->queue->push(JobPayload::immediate(
|
|
new SendWelcomeEmailJob($user->id, $user->email, $user->name)
|
|
));
|
|
|
|
return $user;
|
|
}
|
|
}
|
|
```
|
|
|
|
## Console Commands
|
|
|
|
The framework provides console commands for queue management:
|
|
|
|
```bash
|
|
# Monitor queue status
|
|
php console.php queue:status
|
|
|
|
# Process jobs (worker)
|
|
php console.php queue:work --queue=default --timeout=60
|
|
|
|
# Retry failed jobs
|
|
php console.php queue:retry --queue=default --limit=50
|
|
|
|
# Clear queue
|
|
php console.php queue:clear --queue=default
|
|
|
|
# View failed jobs
|
|
php console.php queue:failed --limit=20
|
|
|
|
# Clean up old jobs
|
|
php console.php queue:cleanup --older-than=7d
|
|
```
|
|
|
|
## Performance Characteristics
|
|
|
|
**Redis Queue**:
|
|
- Throughput: 10,000+ jobs/second
|
|
- Latency: <5ms per operation
|
|
- Memory: ~200 bytes per job (serialized)
|
|
- Scalability: Horizontal (add Redis instances)
|
|
|
|
**File Queue**:
|
|
- Throughput: 1,000+ jobs/second
|
|
- Latency: <20ms per operation
|
|
- Memory: Minimal (disk-based)
|
|
- Scalability: Vertical (single server)
|
|
|
|
**Job Execution**:
|
|
- Overhead: <10ms per job (framework overhead)
|
|
- Timeout: Configurable per job
|
|
- Memory: Depends on job complexity
|
|
- Concurrency: Multiple workers supported |