queue = new VQueue(); $this->logger = VLogger::getInstance(); // Handle shutdown signals gracefully if (function_exists('pcntl_signal')) { pcntl_signal(SIGTERM, [$this, 'shutdown']); pcntl_signal(SIGINT, [$this, 'shutdown']); } } /** * Start processing jobs from specified queues * @param array $queues Queue names to process */ public function work($queues = ['default', 'video_processing', 'notifications']) { $this->logger->info('Queue worker started', [ 'queues' => $queues, 'pid' => getmypid() ]); while ($this->running && $this->jobsProcessed < $this->maxJobs) { try { // Check for signals if (function_exists('pcntl_signal_dispatch')) { pcntl_signal_dispatch(); } // Get next job $job = $this->queue->dequeue($queues, 10); // 10 second timeout if ($job) { $this->processJob($job); $this->jobsProcessed++; } else { // No jobs available, sleep briefly usleep(100000); // 0.1 seconds } // Memory cleanup if ($this->jobsProcessed % 10 === 0) { gc_collect_cycles(); } } catch (Exception $e) { $this->logger->error('Queue worker error', [ 'error' => $e->getMessage(), 'trace' => $e->getTraceAsString() ]); // Sleep on error to prevent rapid error loops sleep(5); } } $this->logger->info('Queue worker stopped', [ 'jobs_processed' => $this->jobsProcessed, 'reason' => $this->running ? 'max_jobs_reached' : 'shutdown_signal' ]); } /** * Process a single job * @param array $job Job data */ private function processJob($job) { $startTime = microtime(true); $this->logger->info('Processing job', [ 'job_id' => $job['id'], 'class' => $job['class'], 'queue' => $job['queue'], 'attempt' => $job['attempts'] ]); try { // Load job class $jobClass = $job['class']; if (!$this->loadJobClass($jobClass)) { throw new Exception("Job class {$jobClass} not found"); } // Create job instance and execute $jobInstance = new $jobClass(); if (!method_exists($jobInstance, 'handle')) { throw new Exception("Job class {$jobClass} must have a handle method"); } // Execute job $result = $jobInstance->handle($job['data']); // Mark job as completed $this->queue->markCompleted($job['id'], $result); $processingTime = microtime(true) - $startTime; $this->logger->info('Job completed successfully', [ 'job_id' => $job['id'], 'class' => $job['class'], 'processing_time' => $processingTime, 'memory_usage' => memory_get_usage(true) ]); } catch (Exception $e) { $processingTime = microtime(true) - $startTime; $this->logger->error('Job failed', [ 'job_id' => $job['id'], 'class' => $job['class'], 'error' => $e->getMessage(), 'processing_time' => $processingTime, 'attempt' => $job['attempts'] ]); // Mark job as failed (queue system will handle retries) $this->queue->markFailed($job['id'], $e->getMessage()); } } /** * Shutdown handler */ public function shutdown() { $this->logger->info('Queue worker shutdown signal received'); $this->running = false; } /** * Attempt to load a queue job class from supported directories * @param string $jobClass * @return bool */ private function loadJobClass($jobClass) { if (class_exists($jobClass)) { return true; } $rootDir = dirname(__DIR__, 2); $searchDirectories = [ __DIR__, $rootDir . DIRECTORY_SEPARATOR . 'f_jobs', $rootDir . DIRECTORY_SEPARATOR . 'f_core' . DIRECTORY_SEPARATOR . 'f_workers', ]; foreach ($searchDirectories as $directory) { $jobFile = rtrim($directory, DIRECTORY_SEPARATOR) . DIRECTORY_SEPARATOR . $jobClass . '.php'; if (!is_file($jobFile)) { continue; } require_once $jobFile; if (class_exists($jobClass)) { if (method_exists($this->logger, 'debug')) { $this->logger->debug('Job class loaded', [ 'class' => $jobClass, 'path' => $jobFile ]); } return true; } } return class_exists($jobClass); } } // CLI execution if (php_sapi_name() === 'cli') { $worker = new QueueWorker(); // Get queues from command line arguments $queues = array_slice($argv, 1); if (empty($queues)) { $queues = ['default', 'video_processing', 'notifications']; } echo "Starting queue worker for queues: " . implode(', ', $queues) . "\n"; echo "Press Ctrl+C to stop\n\n"; $worker->work($queues); }