How to Use Queue To Speedup PHP Processing Tasks Part 2: Processing Queued Tasks

<?php

class Worker {

 protected $_queue;

 /**
  * remember the queue reference
  * @param object $queue
  */
 public function __construct($queue) {
  $this->_queue = $queue;
 }

 public function process() {
   $item = $this->_queue->getItem();
   manipulate_the_item_to_do_the_work($item);
   return true;
 }
};

$worker->process();

?>

CREATE TABLE IF NOT EXISTS `queue` (
  `id` int(10) NOT NULL PRIMARY KEY AUTO_INCREMENT,
  `email` varchar(100) NOT NULL,
  `html` text NOT NULL,
  `status` enum('queued','processing','done','failed') NOT NULL DEFAULT 'queued',
  `error_text` varchar(255) DEFAULT NULL,
  `created_at` datetime DEFAULT NULL,
  `updated_at` timestamp NOT NULL 
        DEFAULT '0000-00-00 00:00:00'
        ON UPDATE CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

• status field equals "queued" value by default— that means that a job is queued and is waiting for the processing. When we take an element from the queue to process, we change this field value to "processing". If the processing was successful, we set this value to "done", otherwise to "failed" and log the error message.

• From my experience it is best to log the timestamp when the item was queued and the timestamp of the last status change. This is saved in created_at and updated_at fields accordingly. When we just create a new record in this table, we set created_at to equal to current time. And every time the record is updated (say, we change the status field), the updated_at  field gets updated automatically — it's a trick of MySQL for TIMESTAMP fields that have "ON UPDATE CURRENT_TIMESTAMP" setting.

Queuing Tasks

<?php

class Queue {

 protected $_databaseHandler;

 /**
  * remember the database reference
  * @param object $databaseHandler
  */
 public function __construct($databaseHandler) {
  $this->_databaseHandler = $databaseHandler;
 }


 /**
  * runs an INSERT query to append a new task to the existing queue table
  * @param string $emailAddress
  * @param string $html
  */
 public function addItem($emailAddress, $html) {
    $this->_databaseHandler->query('INSERT INTO queue
       (`id`, `email`, `html`, `created_at`, `updated_at`) 
       VALUES (NULL, "' . quote($emailAddress). '",  "'
          . quote($html). '", NOW(), NOW())';
    return true;
 }


 /**
  * requests a new queued item (not being processed, done or failed. A new one!)
  * @return object $item
  */
 public function getItem() {
  $item = $this->_databaseHandler->query('SELECT * 
      FROM queue
      WHERE status="queued"
      LIMIT 1');
  return $item;
 }
 
};

?>

<?php

class Worker {

 protected $_databaseHandler;
 protected $_queue;

 /**
  * remember the references
  * @param object $queue
  * @param object $databaseHandler
  */
 public function __construct($queue, $databaseHandler) {
  $this->_queue = $queue;
  $this->_databaseHandler = $databaseHandler;
 }

 public function process() {
   $item = $this->_queue->getItem();
   $this->markItemAsBeingProcessed($item);

   try {
     mail(
       $item['email'],
       'Your are subscribed!',
       $item['html']
     );

     // if mail() was ok, just mark the item as done
     $this->markItemAsDone($item);
   } 
   catch (Exception $e) {
     //if mail() fails, mark item as failed and save the error text
     $errorMessageText = $e->getMessage();
     $this->markItemAsFailed($item, $errorMessageText);
   }
 }

 public function markItemAsBeingProcessed($item) {
   $this->_changeItemStatus($item['id'], 'processing');
 }

 public function markItemAsDone($item) {
   $this->_changeItemStatus($item['id'], 'done');
 }

 public function markItemAsFailed($item, $errorMessageText) {
   $this->_changeItemStatus($item['id'], 'failed', $errorMessageText);
 }

 protected function _changeItemStatus($itemId, $itemStatus, $errorMessageText=null) {
    $this->_databaseHandler->query('UPDATE queue SET 
      `status` = '. $itemStatus . ',
      `error_text` = '. $errorMessageText . '
      WHERE `id` = '. $itemId;
 }
 
};

?>