php-beanstalkd消息队列类实例分享
程序员文章站
2024-03-31 20:30:28
本文实例为大家分享了php beanstalkd消息队列类的具体代码,供大家参考,具体内容如下
本文实例为大家分享了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; } } ?>
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持。