本專案實現基於beanstalk的php訊息佇列服務，包括生產與消費訊息案例

<?php


/**
 * 訊息生產/接收類
 * @example
 *  // 生產單條訊息,goods管道組
 *  $mq = new MessageQueue();
 *  $mq->product('goods', 111111);
 *  // 生產多條訊息
 *  $mq->product('goods', array(111111, 111112));
 *
 *  // 訊息佇列監聽處理指令碼,goods管道組,solr管道
 *  <?php
 *      $mq = new MessageQueue('solr');
 *      $mq->watch('goods', function($message) {
 *          $goods_id = intval($message);
 *          // 以下為具體業務處理邏輯
 *          // 
 *          // ...
 *          // 返回true表示已處理完畢，伺服器將刪除該條訊息
 *          return true;
 *      });
 */

require_once dirname(__DIR__).'/BeanstalkClient.php';

class MessageQueue {

    // 訂閱者ID
    private $_clientID = null;

    // 訂閱者清單
    private $_subscribers = array();

    // beanstalkd連線配置資訊
    private $_beanstalkdConfig = array();

    /**
     * beanstalk client
     * @var BeanstalkClient
     */
    private $_beanstalk = null;

    /**
     * 初始化訊息客戶端
     * @param string $clientID 分配給訊息接受端的ID標識
     */
    public function __construct($clientID = null) {
        $this->_clientID = $clientID;
        $this->_setConfig();
    }

    /**
     * 生產訊息, 對管道內的所有事件推送訊息
     * @param string $queue 佇列名 -- 管道組
     * @param [string|array] $messages 訊息內容，多條使用陣列
     */
    public function product($queue, $messages) {
        try {
            if(!isset($this->_subscribers[$queue])) {
                throw new Exception('queue of "'.$queue.'" havn\'t configured, '
                    .'go '.__DIR__.'/../config/params.php and configure it');
            }
            $beanstalk = $this->getBeanstalkClient();
            if(!is_array($messages)) {
                $messages = array($messages);
            }
            foreach($this->_subscribers[$queue] as $clientID) {
                $beanstalk->useTube($queue.'.'.$clientID);
                foreach($messages as $message) {
                	if(strlen($message)){
                		$beanstalk->put(11, 0, 60, $message);
                		$this->_log('product', $queue, $clientID, $message);
                	}
                }
            }
        } catch (Exception $e) {
            throw new Exception($e->getMessage());
        }
        return true;
    }

    /**
     * 根據佇列名和事件名投遞訊息, 只對指定管道和事件推送訊息
     * @param  [string] $queue   [佇列名]
     * @param  [string] $event   [事件名]
     * @param  [string] $message [訊息內容]
     * @param  [int] $delay [延時時間]
     * @return [void]
     */
    public function product_conf($queue, $event, $message, $delay = 0) {
        $beanstalk = $this->getBeanstalkClient();
        $beanstalk->useTube("{$queue}.{$event}");
        $beanstalk->put(11, $delay, 60, $message);
    }

    /**
     * 監聽佇列並處理訊息
     * @param string $queue 訂閱的佇列名
     * @param function $callback 回撥方法(訊息處理函式，會將訊息內容作為引數給$callback)
     */
    public function watch($queue, $callback) {
        try {

            if(!$this->_checkQueueExist($queue, $this->_clientID)) {
            	$this->_log('checkQueue', $queue, $this->_clientID, '');
                throw new Exception($this->_clientID.' is not allow to access this queue');
            }
            if(!is_object($callback)) {
            	$this->_log('isObject', $queue, $this->_clientID,'');
                throw new Exception('param of callback is not a function');
            }
            $this->_beanstalkdConfig['persistent'] = false;
            $beanstalk = new BeanstalkClient($this->_beanstalkdConfig);
            $beanstalk->connect();
            $beanstalk->watch($queue.'.'.$this->_clientID);
            $retry = 0;
            for(;;) {
                $job = $beanstalk->reserve();
                if($job) {
                    $result = $callback($job['body']);
                    //處理任務
                    if(true === $result) {
                        $beanstalk->delete($job['id']);
                        $this->_log('consume', $queue, $this->_clientID, $job['body']);
                    }else{
                        $beanstalk->bury($job['id'],'');
                        $this->_log('bury', $queue, $this->_clientID, $job['body']);
                    }
                } else {
                    $this->_log('error', $queue, $this->_clientID, $job['body']);
                    // 設定 error_reporting(0) 時watcher指令碼會陷入死迴圈，這裡設定重連
                    if ($retry++ >= 10) {
                      $retry = 0;
                      $this->_log('error', $queue, $this->_clientID, 'try to reconnect.');
                        sleep(5); // 等待beanstalkd服務恢復
                        $beanstalk->connect();
                        $beanstalk->watch($queue.'.'.$this->_clientID);
                    }
                }
            }
            $beanstalk->disconnect();
        } catch (Exception $e) {
        	$this->_log('error', $queue, $this->_clientID,'');
            throw new Exception($e->getMessage());
        }
    }


    /**
     * 初始化配置資訊
     */
    private function _setConfig() {
        $config = require dirname(__DIR__).'/config/params.php';
        $this->_subscribers = $config['subscribers'];
        $this->_beanstalkdConfig = $config['beanstalkd'];
    }

    /**
     * 檢查當前客戶端監聽的佇列是否存在
     * @param string $queue 佇列名
     * @param string $clientID 客戶端ID
     * @return boolean
     */
    private function _checkQueueExist($queue, $clientID) {
        return isset($this->_subscribers[$queue]) && in_array($clientID, $this->_subscribers[$queue]);
    }

    /**
     * 獲取beanstalk client
     * @param  array  $config 連線配置
     * @return BeanstalkClient
     */
    private function getBeanstalkClient()
    {
        if (is_null($this->_beanstalk)) {
            $this->_beanstalk = new BeanstalkClient($this->_beanstalkdConfig);
            $this->_beanstalk->connect();
        }
        try {
            // 檢查連線
            $this->_beanstalk->stats();
        } catch (Exception $e) {
            // 若出錯則重連
            $this->_beanstalk->connect();
        }
        return $this->_beanstalk;
    }

    /**
     * 記錄日誌
     * @param string $operation 操作型別
     * @param string $queue 佇列名 -- 管道組
     * @param string $clientID 客戶端ID -- 管道
     * @param string $message 訊息體
     */
    public function _log($operation, $queue, $clientID, $message, $folder = 'mqlog') {
    	$dir = MQ_LOG_PATH;
        (file_exists($dir) && is_dir($dir)) || mkdir($dir, 0777, true);
        $file = $dir.'/'.$queue.'.'.$clientID.'.log';
        $mode = (is_file($file) && filesize($file)/1024/1024 < 20) ? "ab+" : "wb"; // 日誌大於20M則清空, 微分銷系統穩定之前先手動清log
        $fp = fopen($file , $mode);
        if(flock($fp , LOCK_EX)){
            fwrite($fp , '['.date('Y-m-d H:i:s').'] '.$operation.': '.$message.PHP_EOL);
            flock($fp , LOCK_UN);
            @chmod($file, 0777);
        }
        fclose($fp);
    }

    /**
     * destruct
     * disconnect
     */
    public function __destruct()
    {
        if (!is_null($this->_beanstalk)) {
            $this->_beanstalk->disconnect();
        }
    }
}