概況：
Beanstalkd，一個(gè)高性能、輕量級(jí)的分布式內(nèi)存隊(duì)列系統(tǒng)，最初設(shè)計(jì)的目的是想通過后臺(tái)異步執(zhí)行耗時(shí)的任務(wù)來降低高容量Web應(yīng)用系統(tǒng)的頁面訪問延遲，支持過有9.5 million用戶的Facebook Causes應(yīng)用。后來開源，現(xiàn)在有PostRank大規(guī)模部署和使用，每天處理百萬級(jí)任務(wù)。Beanstalkd是典型的類Memcached設(shè)計(jì)，協(xié)議和使用方式都是同樣的風(fēng)格，所以使用過memcached的用戶會(huì)覺得Beanstalkd似曾相識(shí)。
beanstalk核心概念：
job:一個(gè)需要異步處理的任務(wù)，需要放在一個(gè)tube中。
tube:一個(gè)有名的任務(wù)隊(duì)列，用來存儲(chǔ)統(tǒng)一類型的job
producer:job的生產(chǎn)者
consumer:job的消費(fèi)者
簡單來說流程就一句話：
由 producer 產(chǎn)生一個(gè)任務(wù) job ，并將 job 推進(jìn)到一個(gè) tube 中，
然后由 consumer 從 tube 中取出 job 執(zhí)行（當(dāng)然了，這一切的操作的前提是beanstalk服務(wù)正在運(yùn)行中）。

一個(gè)job有READY（時(shí)刻準(zhǔn)備著被消費(fèi)者取出）, RESERVED（任務(wù)正在被一個(gè)消費(fèi)者處理中）, DELAYED（延遲任務(wù)，設(shè)定的延遲時(shí)間后進(jìn)入ready狀態(tài)）, BURIED（休眠中，需要轉(zhuǎn)移狀態(tài)后才能操作）四種狀態(tài)。當(dāng)producer直接put一個(gè)job時(shí)，job就處于READY狀態(tài)，等待consumer來處理，如果選擇延遲put，job就先到DELAYED狀態(tài)，等待時(shí)間過后才遷移到READY狀態(tài)。consumer獲取了當(dāng)前READY的job后，該job的狀態(tài)就遷移到RESERVED，這樣其他的consumer就不能再操作該job。當(dāng)consumer完成該job后，可以選擇delete, release或者bury操作;delete之后，job從系統(tǒng)消亡，之后不能再獲取;release操作可以重新把該job狀態(tài)遷移回READY(也可以延遲該狀態(tài)遷移操作)，使其他的consumer可以繼續(xù)獲取和執(zhí)行該job;有意思的是bury操作，可以把該job休眠，等到需要的時(shí)候，再將休眠的job kick回READY狀態(tài)，也可以delete BURIED狀態(tài)的job。正是有這些有趣的操作和狀態(tài)，才可以基于此做出很多意思的應(yīng)用，比如要實(shí)現(xiàn)一個(gè)循環(huán)隊(duì)列，就可以將RESERVED狀態(tài)的job休眠掉，等沒有READY狀態(tài)的job時(shí)再將BURIED狀態(tài)的job一次性kick回READY狀態(tài)。

例子分析：微博是一個(gè)很典型的例子：
1，發(fā)一個(gè)微博
2，推送給他的粉絲 （如果有100w個(gè)粉絲，這個(gè)地方會(huì)堵塞很久，用戶感受到的就是延遲）
在微博上發(fā)布一條內(nèi)容要做上面兩件事情才算完整，發(fā)一條微博只需要進(jìn)行一次簡單的數(shù)據(jù)庫操作，
但是推送給他的粉絲卻要操作100w次數(shù)據(jù)庫，導(dǎo)致用戶發(fā)一個(gè)微博要等待很長的延遲才能返回結(jié)果發(fā)送成功。
采用隊(duì)列的方式，用戶發(fā)送一條微博立馬返回結(jié)果，發(fā)送成功，剩下的推送就放到隊(duì)列里面異步執(zhí)行，
推送并不需要特別及時(shí)，延遲過幾秒幾十秒都是可以接受的。

本文實(shí)例為大家分享了php beanstalkd消息隊(duì)列類的具體代碼，供大家參考，具體內(nèi)容如下

<?phpnamespace Common/Business;/** * beanstalk: A minimalistic PHP beanstalk client. * * Copyright (c) 2009-2015 David Persson * * Distributed under the terms of the MIT License. * Redistributions of files must retain the above copyright notice. */ use RuntimeException; /** * An interface to the beanstalk queue service. Implements the beanstalk * protocol spec 1.9. Where appropriate the documentation from the protocol * has been added to the docblocks in this html' target='_blank'>class. * * @link https://github.com/kr/beanstalkd/blob/master/doc/protocol.txt */class BeanStalk {   /**   * Minimum priority value which can be assigned to a job. The minimum   * priority value is also the _highest priority_ a job can have.   *   * @var integer   */  const MIN_PRIORITY = 0;   /**   * Maximum priority value which can be assigned to a job. The maximum   * priority value is also the _lowest priority_ a job can have.   *   * @var integer   */  const MAX_PRIORITY = 4294967295;   /**   * Holds a boolean indicating whether a connection to the server is   * currently established or not.   *   * @var boolean   */  public $connected = false;   /**   * Holds configuration values.   *   * @var array   */  protected $_config = [];   /**   * The current connection resource handle (if any).   *   * @var resource   */  protected $_connection;   /**   * Constructor.   *   * @param array $config An array of configuration values:   *    - `'persistent'` Whether to make the connection persistent or   *             not, defaults to `true` as the FAQ recommends   *             persistent connections.   *    - `'host'`    The beanstalk server hostname or IP address to   *             connect to, defaults to `127.0.0.1`.   *    - `'port'`    The port of the server to connect to, defaults   *             to `11300`.   *    - `'timeout'`   Timeout in seconds when establishing the   *             connection, defaults to `1`.   *    - `'logger'`   An instance of a PSR-3 compatible logger.   *   * @link https://github.com/php-fig/fig-standards/blob/master/accepted/PSR-3-logger-interface.md   * @return void   */  public function __construct(array $config = []) {    $defaults = [      'persistent' => true,      'host' => '127.0.0.1',      'port' => 11300,      'timeout' => 1,      'logger' => null    ];    $this->_config = $config + $defaults;  }   /**   * Destructor, disconnects from the server.   *   * @return void   */  public function __destruct() {    $this->disconnect();  }   /**   * Initiates a socket connection to the beanstalk server. The resulting   * stream will not have any timeout set on it. Which means it can wait   * an unlimited amount of time until a packet becomes available. This   * is required for doing blocking reads.   *   * @see /Beanstalk/Client::$_connection   * @see /Beanstalk/Client::reserve()   * @return boolean `true` if the connection was established, `false` otherwise.   */  public function connect() {    if (isset($this->_connection)) {      $this->disconnect();    }    $errNum = '';    $errStr = '';    $function = $this->_config['persistent'] ? 'pfsockopen' : 'fsockopen';    $params = [$this->_config['host'], $this->_config['port'], &$errNum, &$errStr];     if ($this->_config['timeout']) {      $params[] = $this->_config['timeout'];    }    $this->_connection = @call_user_func_array($function, $params);     if (!empty($errNum) || !empty($errStr)) {      $this->_error("{$errNum}: {$errStr}");    }     $this->connected = is_resource($this->_connection);     if ($this->connected) {      stream_set_timeout($this->_connection, -1);    }    return $this->connected;  }   /**   * Closes the connection to the beanstalk server by first signaling   * that we want to quit then actually closing the socket connection.   *   * @return boolean `true` if diconnecting was successful.   */  public function disconnect() {    if (!is_resource($this->_connection)) {      $this->connected = false;    } else {      $this->_write('quit');      $this->connected = !fclose($this->_connection);       if (!$this->connected) {        $this->_connection = null;      }    }    return !$this->connected;  }   /**   * Pushes an error message to the logger, when one is configured.   *   * @param string $message The error message.   * @return void   */  protected function _error($message) {    if ($this->_config['logger']) {      $this->_config['logger']->error($message);    }  }   public function errors()  {    return $this->_config['logger'];  }  /**   * Writes a packet to the socket. Prior to writing to the socket will   * check for availability of the connection.   *   * @param string $data   * @return integer|boolean number of written bytes or `false` on error.   */  protected function _write($data) {    if (!$this->connected) {      $message = 'No connecting found while writing data to socket.';      throw new RuntimeException($message);    }     $data .= "/r/n";    return fwrite($this->_connection, $data, strlen($data));  }   /**   * Reads a packet from the socket. Prior to reading from the socket   * will check for availability of the connection.   *   * @param integer $length Number of bytes to read.   * @return string|boolean Data or `false` on error.   */  protected function _read($length = null) {    if (!$this->connected) {      $message = 'No connection found while reading data from socket.';      throw new RuntimeException($message);    }    if ($length) {      if (feof($this->_connection)) {        return false;      }      $data = stream_get_contents($this->_connection, $length + 2);      $meta = stream_get_meta_data($this->_connection);       if ($meta['timed_out']) {        $message = 'Connection timed out while reading data from socket.';        throw new RuntimeException($message);      }      $packet = rtrim($data, "/r/n");    } else {      $packet = stream_get_line($this->_connection, 16384, "/r/n");    }    return $packet;  }   /* Producer Commands */   /**   * The `put` command is for any process that wants to insert a job into the queue.   *   * @param integer $pri Jobs with smaller priority values will be scheduled   *    before jobs with larger priorities. The most urgent priority is   *    0; the least urgent priority is 4294967295.   * @param integer $delay Seconds to wait before putting the job in the   *    ready queue. The job will be in the "delayed" state during this time.   * @param integer $ttr Time to run - Number of seconds to allow a worker to   *    run this job. The minimum ttr is 1.   * @param string $data The job body.   * @return integer|boolean `false` on error otherwise an integer indicating   *     the job id.   */  public function put($pri, $delay, $ttr, $data) {    $this->_write(sprintf("put %d %d %d %d/r/n%s", $pri, $delay, $ttr, strlen($data), $data));    $status = strtok($this->_read(), ' ');     switch ($status) {      case 'INSERTED':      case 'BURIED':        return (integer) strtok(' '); // job id      case 'EXPECTED_CRLF':      case 'JOB_TOO_BIG':      default:        $this->_error($status);        return false;    }  }   /**   * The `use` command is for producers. Subsequent put commands will put   * jobs into the tube specified by this command. If no use command has   * been issued, jobs will be put into the tube named `default`.   *   * @param string $tube A name at most 200 bytes. It specifies the tube to   *    use. If the tube does not exist, it will be created.   * @return string|boolean `false` on error otherwise the name of the tube.   */  public function useTube($tube) {    $this->_write(sprintf('use %s', $tube));    $status = strtok($this->_read(), ' ');     switch ($status) {      case 'USING':        return strtok(' ');      default:        $this->_error($status);        return false;    }  }   /**   * Pause a tube delaying any new job in it being reserved for a given time.   *   * @param string $tube The name of the tube to pause.   * @param integer $delay Number of seconds to wait before reserving any more   *    jobs from the queue.   * @return boolean `false` on error otherwise `true`.   */  public function pauseTube($tube, $delay) {    $this->_write(sprintf('pause-tube %s %d', $tube, $delay));    $status = strtok($this->_read(), ' ');     switch ($status) {      case 'PAUSED':        return true;      case 'NOT_FOUND':      default:        $this->_error($status);        return false;    }  }   /* Worker Commands */   /**   * Reserve a job (with a timeout).   *   * @param integer $timeout If given specifies number of seconds to wait for   *    a job. `0` returns immediately.   * @return array|false `false` on error otherwise an array holding job id   *     and body.   */  public function reserve($timeout = null) {    if (isset($timeout)) {      $this->_write(sprintf('reserve-with-timeout %d', $timeout));    } else {      $this->_write('reserve');    }    $status = strtok($this->_read(), ' ');     switch ($status) {      case 'RESERVED':        return [          'id' => (integer) strtok(' '),          'body' => $this->_read((integer) strtok(' '))        ];      case 'DEADLINE_SOON':      case 'TIMED_OUT':      default:        $this->_error($status);        return false;    }  }   /**   * Removes a job from the server entirely.   *   * @param integer $id The id of the job.   * @return boolean `false` on error, `true` on success.   */  public function delete($id) {    $this->_write(sprintf('delete %d', $id));    $status = $this->_read();     switch ($status) {      case 'DELETED':        return true;      case 'NOT_FOUND':      default:        $this->_error($status);        return false;    }  }   /**   * Puts a reserved job back into the ready queue.   *   * @param integer $id The id of the job.   * @param integer $pri Priority to assign to the job.   * @param integer $delay Number of seconds to wait before putting the job in the ready queue.   * @return boolean `false` on error, `true` on success.   */  public function release($id, $pri, $delay) {    $this->_write(sprintf('release %d %d %d', $id, $pri, $delay));    $status = $this->_read();     switch ($status) {      case 'RELEASED':      case 'BURIED':        return true;      case 'NOT_FOUND':      default:        $this->_error($status);        return false;    }  }   /**   * Puts a job into the `buried` state Buried jobs are put into a FIFO   * linked list and will not be touched until a client kicks them.   *   * @param integer $id The id of the job.   * @param integer $pri *New* priority to assign to the job.   * @return boolean `false` on error, `true` on success.   */  public function bury($id, $pri) {    $this->_write(sprintf('bury %d %d', $id, $pri));    $status = $this->_read();     switch ($status) {      case 'BURIED':        return true;      case 'NOT_FOUND':      default:        $this->_error($status);        return false;    }  }   /**   * Allows a worker to request more time to work on a job.   *   * @param integer $id The id of the job.   * @return boolean `false` on error, `true` on success.   */  public function touch($id) {    $this->_write(sprintf('touch %d', $id));    $status = $this->_read();     switch ($status) {      case 'TOUCHED':        return true;      case 'NOT_TOUCHED':      default:        $this->_error($status);        return false;    }  }   /**   * Adds the named tube to the watch list for the current connection.   *   * @param string $tube Name of tube to watch.   * @return integer|boolean `false` on error otherwise number of tubes in watch list.   */  public function watch($tube) {    $this->_write(sprintf('watch %s', $tube));    $status = strtok($this->_read(), ' ');     switch ($status) {      case 'WATCHING':        return (integer) strtok(' ');      default:        $this->_error($status);        return false;    }  }   /**   * Remove the named tube from the watch list.   *   * @param string $tube Name of tube to ignore.   * @return integer|boolean `false` on error otherwise number of tubes in watch list.   */  public function ignore($tube) {    $this->_write(sprintf('ignore %s', $tube));    $status = strtok($this->_read(), ' ');     switch ($status) {      case 'WATCHING':        return (integer) strtok(' ');      case 'NOT_IGNORED':      default:        $this->_error($status);        return false;    }  }   /* Other Commands */   /**   * Inspect a job by its id.   *   * @param integer $id The id of the job.   * @return string|boolean `false` on error otherwise the body of the job.   */  public function peek($id) {    $this->_write(sprintf('peek %d', $id));    return $this->_peekRead();  }   /**   * Inspect the next ready job.   *   * @return string|boolean `false` on error otherwise the body of the job.   */  public function peekReady() {    $this->_write('peek-ready');    return $this->_peekRead();  }   /**   * Inspect the job with the shortest delay left.   *   * @return string|boolean `false` on error otherwise the body of the job.   */  public function peekDelayed() {    $this->_write('peek-delayed');    return $this->_peekRead();  }   /**   * Inspect the next job in the list of buried jobs.   *   * @return string|boolean `false` on error otherwise the body of the job.   */  public function peekBuried() {    $this->_write('peek-buried');    return $this->_peekRead();  }   /**   * Handles response for all peek methods.   *   * @return string|boolean `false` on error otherwise the body of the job.   */  protected function _peekRead() {    $status = strtok($this->_read(), ' ');     switch ($status) {      case 'FOUND':        return [          'id' => (integer) strtok(' '),          'body' => $this->_read((integer) strtok(' '))        ];      case 'NOT_FOUND':      default:        $this->_error($status);        return false;    }  }   /**   * Moves jobs into the ready queue (applies to the current tube).   *   * If there are buried jobs those get kicked only otherwise delayed   * jobs get kicked.   *   * @param integer $bound Upper bound on the number of jobs to kick.   * @return integer|boolean False on error otherwise number of jobs kicked.   */  public function kick($bound) {    $this->_write(sprintf('kick %d', $bound));    $status = strtok($this->_read(), ' ');     switch ($status) {      case 'KICKED':        return (integer) strtok(' ');      default:        $this->_error($status);        return false;    }  }   /**   * This is a variant of the kick command that operates with a single   * job identified by its job id. If the given job id exists and is in a   * buried or delayed state, it will be moved to the ready queue of the   * the same tube where it currently belongs.   *   * @param integer $id The job id.   * @return boolean `false` on error `true` otherwise.   */  public function kickJob($id) {    $this->_write(sprintf('kick-job %d', $id));    $status = strtok($this->_read(), ' ');     switch ($status) {      case 'KICKED':        return true;      case 'NOT_FOUND':      default:        $this->_error($status);        return false;    }  }   /* Stats Commands */   /**   * Gives statistical information about the specified job if it exists.   *   * @param integer $id The job id.   * @return string|boolean `false` on error otherwise a string with a yaml formatted dictionary.   */  public function statsJob($id) {    $this->_write(sprintf('stats-job %d', $id));    return $this->_statsRead();  }   /**   * Gives statistical information about the specified tube if it exists.   *   * @param string $tube Name of the tube.   * @return string|boolean `false` on error otherwise a string with a yaml formatted dictionary.   */  public function statsTube($tube) {    $this->_write(sprintf('stats-tube %s', $tube));    return $this->_statsRead();  }   /**   * Gives statistical information about the system as a whole.   *   * @return string|boolean `false` on error otherwise a string with a yaml formatted dictionary.   */  public function stats() {    $this->_write('stats');    return $this->_statsRead();  }   /**   * Returns a list of all existing tubes.   *   * @return string|boolean `false` on error otherwise a string with a yaml formatted list.   */  public function listTubes() {    $this->_write('list-tubes');    return $this->_statsRead();  }   /**   * Returns the tube currently being used by the producer.   *   * @return string|boolean `false` on error otherwise a string with the name of the tube.   */  public function listTubeUsed() {    $this->_write('list-tube-used');    $status = strtok($this->_read(), ' ');     switch ($status) {      case 'USING':        return strtok(' ');      default:        $this->_error($status);        return false;    }  }   /**   * Returns a list of tubes currently being watched by the worker.   *   * @return string|boolean `false` on error otherwise a string with a yaml formatted list.   */  public function listTubesWatched() {    $this->_write('list-tubes-watched');    return $this->_statsRead();  }   /**   * Handles responses for all stat methods.   *   * @param boolean $decode Whether to decode data before returning it or not. Default is `true`.   * @return array|string|boolean `false` on error otherwise statistical data.   */  protected function _statsRead($decode = true) {    $status = strtok($this->_read(), ' ');     switch ($status) {      case 'OK':        $data = $this->_read((integer) strtok(' '));        return $decode ? $this->_decode($data) : $data;      default:        $this->_error($status);        return false;    }  }   /**   * Decodes YAML data. This is a super naive decoder which just works on   * a subset of YAML which is commonly returned by beanstalk.   *   * @param string $data The data in YAML format, can be either a list or a dictionary.   * @return array An (associative) array of the converted data.   */  protected function _decode($data) {    $data = array_slice(explode("/n", $data), 1);    $result = [];     foreach ($data as $key => $value) {      if ($value[0] === '-') {        $value = ltrim($value, '- ');      } elseif (strpos($value, ':') !== false) {        list($key, $value) = explode(':', $value);        $value = ltrim($value, ' ');      }      if (is_numeric($value)) {        $value = (integer) $value == $value ? (integer) $value : (float) $value;      }      $result[$key] = $value;    }    return $result;  }} ?>

以上就是關(guān)于php中beanstalkd消息隊(duì)列的詳解以及類的分享的詳細(xì)內(nèi)容，更多請關(guān)注 其它相關(guān)文章！

鄭重聲明：本文版權(quán)歸原作者所有，轉(zhuǎn)載文章僅為傳播更多信息之目的，如作者信息標(biāo)記有誤，請第一時(shí)間聯(lián)系我們修改或刪除，多謝。