PHPでのBeantalkdメッセージキューとクラスの共有の詳細な説明

この記事では、主に php-beanstalkd メッセージ キュー クラスのサンプル コードを共有します。これには特定の参考値があります。興味のある方は参照してください。

概要:

Beanstalkd は、高性能で軽量な分散型です。時間のかかるタスクをバックグラウンドで非同期に実行することで、大容量の Web アプリケーション システムのページ アクセスの遅延を軽減し、950 万人のユーザーを抱える Facebook Causes アプリケーションをサポートしました。その後、PostRank はオープンソース化され、現在では大規模に展開および使用され、毎日何百万ものタスクを処理しています。 Beanstalkd は典型的な Memcached に似た設計であり、プロトコルと使用法は同じであるため、memcached を使用したことのあるユーザーは Beanstalkd に親しみを感じるでしょう。
beanstalk のコアコンセプト:
job: 非同期処理を必要とするタスクはチューブに配置する必要があります。
チューブ: 統一タイプのジョブを保存するために使用されるよく知られたタスクキュー
プロデューサー: ジョブのプロデューサー
コンシューマー: ジョブのコンシューマー
簡単に言えば、プロセスはたったの 1 つの文です:
プロデューサーは、タスク ジョブとそのジョブの書き込み チューブにプッシュすると、コンシューマーはチューブからジョブを取得して実行します (もちろん、これらすべての操作の前提条件は、Beantalk サービスが実行されていることです)。

ジョブには、READY (常にコンシューマによって取り出せる準備ができている)、RESERVED (タスクがコンシューマによって処理されている)、DELAYED (タスクを遅延させ、設定された遅延時間後に準備完了状態になる)、BURIED (スリープ中) があります。 、必須 状態が転送された後でのみ操作を実行できます） 4 つの状態。プロデューサがジョブを直接配置すると、ジョブは READY 状態になり、コンシューマによる処理を待機します。配置を遅らせることを選択した場合、ジョブはまず DELAYED 状態になり、その後 READY 状態に移行します。待ち時間。コンシューマが現在の READY ジョブを取得すると、ジョブのステータスは RESERVED に移行され、他のコンシューマはジョブを操作できなくなります。コンシューマーはジョブの完了後、削除、解放、または埋め込み操作を選択できます。削除後、ジョブはシステムから削除され、解放操作によってジョブのステータスが READY に戻る可能性があります。これにより、他のコンシューマーは引き続きジョブを取得して実行できるようになります。興味深いのは、ジョブをスリープ状態にして、必要に応じて休止状態のジョブを READY 状態に戻すか、削除できることです。ジョブは BURIED 状態になります。これらの興味深い操作と状態があるからこそ、それらに基づいて多くの興味深いアプリケーションを作成できます。たとえば、循環キューを実装したい場合は、ジョブを RESERVED 状態でスリープさせてから、BURIED 状態のジョブを配置することができます。 READY 状態のジョブがないときに動作するように、ジョブは一度に READY 状態に戻ります。

分析例: Weibo は非常に典型的な例です:

1. Weibo を投稿する

2. ファンにプッシュする (100 万人のファンがいる場合、この場所は長時間ブロックされ、ユーザーは遅れを感じるでしょう)
Weibo にコンテンツを投稿するには、上記 2 つのことを完了する必要があります。Weibo を投稿するには、簡単なデータベース操作が 1 回だけ必要です。
しかし、それをプッシュするファンはデータベースを 100 万回操作する必要があるため、ユーザーは投稿を行う必要があります。 a Weibo 結果が正常に返されるまでに長い時間がかかります。
キューメソッドを使用して、ユーザーはWeiboメッセージを送信し、メッセージが正常に送信された場合、残りのプッシュは非同期実行のためにキューに入れられます。
プッシュは特にタイムリーである必要はありません。 、数秒または数十秒の遅延は許容されます。

この記事の例では、参考のために php beantalkd メッセージ キュー クラスの特定のコードを共有します。具体的な内容は次のとおりです

<?php
namespace 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 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:
   *    - `&#39;persistent&#39;` Whether to make the connection persistent or
   *             not, defaults to `true` as the FAQ recommends
   *             persistent connections.
   *    - `&#39;host&#39;`    The beanstalk server hostname or IP address to
   *             connect to, defaults to `127.0.0.1`.
   *    - `&#39;port&#39;`    The port of the server to connect to, defaults
   *             to `11300`.
   *    - `&#39;timeout&#39;`   Timeout in seconds when establishing the
   *             connection, defaults to `1`.
   *    - `&#39;logger&#39;`   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 = [
      &#39;persistent&#39; => true,
      &#39;host&#39; => &#39;127.0.0.1&#39;,
      &#39;port&#39; => 11300,
      &#39;timeout&#39; => 1,
      &#39;logger&#39; => 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 = &#39;&#39;;
    $errStr = &#39;&#39;;
    $function = $this->_config[&#39;persistent&#39;] ? &#39;pfsockopen&#39; : &#39;fsockopen&#39;;
    $params = [$this->_config[&#39;host&#39;], $this->_config[&#39;port&#39;], &$errNum, &$errStr];
 
    if ($this->_config[&#39;timeout&#39;]) {
      $params[] = $this->_config[&#39;timeout&#39;];
    }
    $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(&#39;quit&#39;);
      $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[&#39;logger&#39;]) {
      $this->_config[&#39;logger&#39;]->error($message);
    }
  }
 
  public function errors()
  {
    return $this->_config[&#39;logger&#39;];
  }
  /**
   * 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 = &#39;No connecting found while writing data to socket.&#39;;
      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 = &#39;No connection found while reading data from socket.&#39;;
      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[&#39;timed_out&#39;]) {
        $message = &#39;Connection timed out while reading data from socket.&#39;;
        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(), &#39; &#39;);
 
    switch ($status) {
      case &#39;INSERTED&#39;:
      case &#39;BURIED&#39;:
        return (integer) strtok(&#39; &#39;); // job id
      case &#39;EXPECTED_CRLF&#39;:
      case &#39;JOB_TOO_BIG&#39;:
      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(&#39;use %s&#39;, $tube));
    $status = strtok($this->_read(), &#39; &#39;);
 
    switch ($status) {
      case &#39;USING&#39;:
        return strtok(&#39; &#39;);
      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(&#39;pause-tube %s %d&#39;, $tube, $delay));
    $status = strtok($this->_read(), &#39; &#39;);
 
    switch ($status) {
      case &#39;PAUSED&#39;:
        return true;
      case &#39;NOT_FOUND&#39;:
      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(&#39;reserve-with-timeout %d&#39;, $timeout));
    } else {
      $this->_write(&#39;reserve&#39;);
    }
    $status = strtok($this->_read(), &#39; &#39;);
 
    switch ($status) {
      case &#39;RESERVED&#39;:
        return [
          &#39;id&#39; => (integer) strtok(&#39; &#39;),
          &#39;body&#39; => $this->_read((integer) strtok(&#39; &#39;))
        ];
      case &#39;DEADLINE_SOON&#39;:
      case &#39;TIMED_OUT&#39;:
      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(&#39;delete %d&#39;, $id));
    $status = $this->_read();
 
    switch ($status) {
      case &#39;DELETED&#39;:
        return true;
      case &#39;NOT_FOUND&#39;:
      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(&#39;release %d %d %d&#39;, $id, $pri, $delay));
    $status = $this->_read();
 
    switch ($status) {
      case &#39;RELEASED&#39;:
      case &#39;BURIED&#39;:
        return true;
      case &#39;NOT_FOUND&#39;:
      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(&#39;bury %d %d&#39;, $id, $pri));
    $status = $this->_read();
 
    switch ($status) {
      case &#39;BURIED&#39;:
        return true;
      case &#39;NOT_FOUND&#39;:
      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(&#39;touch %d&#39;, $id));
    $status = $this->_read();
 
    switch ($status) {
      case &#39;TOUCHED&#39;:
        return true;
      case &#39;NOT_TOUCHED&#39;:
      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(&#39;watch %s&#39;, $tube));
    $status = strtok($this->_read(), &#39; &#39;);
 
    switch ($status) {
      case &#39;WATCHING&#39;:
        return (integer) strtok(&#39; &#39;);
      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(&#39;ignore %s&#39;, $tube));
    $status = strtok($this->_read(), &#39; &#39;);
 
    switch ($status) {
      case &#39;WATCHING&#39;:
        return (integer) strtok(&#39; &#39;);
      case &#39;NOT_IGNORED&#39;:
      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(&#39;peek %d&#39;, $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(&#39;peek-ready&#39;);
    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(&#39;peek-delayed&#39;);
    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(&#39;peek-buried&#39;);
    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(), &#39; &#39;);
 
    switch ($status) {
      case &#39;FOUND&#39;:
        return [
          &#39;id&#39; => (integer) strtok(&#39; &#39;),
          &#39;body&#39; => $this->_read((integer) strtok(&#39; &#39;))
        ];
      case &#39;NOT_FOUND&#39;:
      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(&#39;kick %d&#39;, $bound));
    $status = strtok($this->_read(), &#39; &#39;);
 
    switch ($status) {
      case &#39;KICKED&#39;:
        return (integer) strtok(&#39; &#39;);
      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(&#39;kick-job %d&#39;, $id));
    $status = strtok($this->_read(), &#39; &#39;);
 
    switch ($status) {
      case &#39;KICKED&#39;:
        return true;
      case &#39;NOT_FOUND&#39;:
      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(&#39;stats-job %d&#39;, $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(&#39;stats-tube %s&#39;, $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(&#39;stats&#39;);
    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(&#39;list-tubes&#39;);
    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(&#39;list-tube-used&#39;);
    $status = strtok($this->_read(), &#39; &#39;);
 
    switch ($status) {
      case &#39;USING&#39;:
        return strtok(&#39; &#39;);
      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(&#39;list-tubes-watched&#39;);
    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(), &#39; &#39;);
 
    switch ($status) {
      case &#39;OK&#39;:
        $data = $this->_read((integer) strtok(&#39; &#39;));
        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] === &#39;-&#39;) {
        $value = ltrim($value, &#39;- &#39;);
      } elseif (strpos($value, &#39;:&#39;) !== false) {
        list($key, $value) = explode(&#39;:&#39;, $value);
        $value = ltrim($value, &#39; &#39;);
      }
      if (is_numeric($value)) {
        $value = (integer) $value == $value ? (integer) $value : (float) $value;
      }
      $result[$key] = $value;
    }
    return $result;
  }
}
 
?>

。

以上がPHPでのBeantalkdメッセージキューとクラスの共有の詳細な説明の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。