这篇文章主要为大家分享了php-beanstalkd消息队列类实例代码，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
概况：
Beanstalkd，一个高性能、轻量级的分布式内存队列系统，最初设计的目的是想通过后台异步执行耗时的任务来降低高容量Web应用系统的页面访问延迟，支持过有9.5 million用户的Facebook Causes应用。后来开源，现在有PostRank大规模部署和使用，每天处理百万级任务。Beanstalkd是典型的类Memcached设计，协议和使用方式都是同样的风格，所以使用过memcached的用户会觉得Beanstalkd似曾相识。
beanstalk核心概念：
job:一个需要异步处理的任务，需要放在一个tube中。
tube:一个有名的任务队列，用来存储统一类型的job
producer:job的生产者
consumer:job的消费者
简单来说流程就一句话：
由 producer 产生一个任务 job ，并将 job 推进到一个 tube 中，
然后由 consumer 从 tube 中取出 job 执行（当然了，这一切的操作的前提是beanstalk服务正在运行中）。
一个job有READY（时刻准备着被消费者取出）, RESERVED（任务正在被一个消费者处理中）, DELAYED（延迟任务，设定的延迟时间后进入ready状态）, BURIED（休眠中，需要转移状态后才能操作）四种状态。当producer直接put一个job时，job就处于READY状态，等待consumer来处理，如果选择延迟put，job就先到DELAYED状态，等待时间过后才迁移到READY状态。consumer获取了当前READY的job后，该job的状态就迁移到RESERVED，这样其他的consumer就不能再操作该job。当consumer完成该job后，可以选择delete, release或者bury操作;delete之后，job从系统消亡，之后不能再获取;release操作可以重新把该job状态迁移回READY(也可以延迟该状态迁移操作)，使其他的consumer可以继续获取和执行该job;有意思的是bury操作，可以把该job休眠，等到需要的时候，再将休眠的job kick回READY状态，也可以delete BURIED状态的job。正是有这些有趣的操作和状态，才可以基于此做出很多意思的应用，比如要实现一个循环队列，就可以将RESERVED状态的job休眠掉，等没有READY状态的job时再将BURIED状态的job一次性kick回READY状态。
例子分析：微博是一个很典型的例子：
1，发一个微博
2，推送给他的粉丝 （如果有100w个粉丝，这个地方会堵塞很久，用户感受到的就是延迟）
在微博上发布一条内容要做上面两件事情才算完整，发一条微博只需要进行一次简单的数据库操作，
但是推送给他的粉丝却要操作100w次数据库，导致用户发一个微博要等待很长的延迟才能返回结果发送成功。
采用队列的方式，用户发送一条微博立马返回结果，发送成功，剩下的推送就放到队列里面异步执行，
推送并不需要特别及时，延迟过几秒几十秒都是可以接受的。
本文实例为大家分享了php beanstalkd消息队列类的具体代码，供大家参考，具体内容如下
<?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:
* - `'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;
}
}
?>Atas ialah kandungan terperinci 关于php中beanstalkd消息队列的详解以及类的分享. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!
Bagaimana untuk membuka fail php
Bagaimana untuk mengalih keluar beberapa elemen pertama tatasusunan dalam php
Apa yang perlu dilakukan jika penyahserialisasian php gagal
Bagaimana untuk menyambungkan php ke pangkalan data mssql
Bagaimana untuk menyambung php ke pangkalan data mssql
Bagaimana untuk memuat naik html
Bagaimana untuk menyelesaikan aksara bercelaru dalam PHP
Bagaimana untuk membuka fail php pada telefon bimbit
![[Web front-end] Permulaan pantas Node.js](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
