本文实例为大家分享了php beanstalkd消息队列类的具体代码,供大家参考,具体内容如下
<"{$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 "\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; } } ?>
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持。
华山资源网 Design By www.eoogi.com
广告合作:本站广告合作请联系QQ:858582 申请时备注:广告合作(否则不回)
免责声明:本站资源来自互联网收集,仅供用于学习和交流,请遵循相关法律法规,本站一切资源不代表本站立场,如有侵权、后门、不妥请联系本站删除!
免责声明:本站资源来自互联网收集,仅供用于学习和交流,请遵循相关法律法规,本站一切资源不代表本站立场,如有侵权、后门、不妥请联系本站删除!
华山资源网 Design By www.eoogi.com
暂无评论...
《魔兽世界》大逃杀!60人新游玩模式《强袭风暴》3月21日上线
暴雪近日发布了《魔兽世界》10.2.6 更新内容,新游玩模式《强袭风暴》即将于3月21 日在亚服上线,届时玩家将前往阿拉希高地展开一场 60 人大逃杀对战。
艾泽拉斯的冒险者已经征服了艾泽拉斯的大地及遥远的彼岸。他们在对抗世界上最致命的敌人时展现出过人的手腕,并且成功阻止终结宇宙等级的威胁。当他们在为即将于《魔兽世界》资料片《地心之战》中来袭的萨拉塔斯势力做战斗准备时,他们还需要在熟悉的阿拉希高地面对一个全新的敌人──那就是彼此。在《巨龙崛起》10.2.6 更新的《强袭风暴》中,玩家将会进入一个全新的海盗主题大逃杀式限时活动,其中包含极高的风险和史诗级的奖励。
《强袭风暴》不是普通的战场,作为一个独立于主游戏之外的活动,玩家可以用大逃杀的风格来体验《魔兽世界》,不分职业、不分装备(除了你在赛局中捡到的),光是技巧和战略的强弱之分就能决定出谁才是能坚持到最后的赢家。本次活动将会开放单人和双人模式,玩家在加入海盗主题的预赛大厅区域前,可以从强袭风暴角色画面新增好友。游玩游戏将可以累计名望轨迹,《巨龙崛起》和《魔兽世界:巫妖王之怒 经典版》的玩家都可以获得奖励。
更新日志
2024年11月17日
2024年11月17日
- 中国武警男声合唱团《辉煌之声1天路》[DTS-WAV分轨]
- 紫薇《旧曲新韵》[320K/MP3][175.29MB]
- 紫薇《旧曲新韵》[FLAC/分轨][550.18MB]
- 周深《反深代词》[先听版][320K/MP3][72.71MB]
- 李佳薇.2024-会发光的【黑籁音乐】【FLAC分轨】
- 后弦.2012-很有爱【天浩盛世】【WAV+CUE】
- 林俊吉.2012-将你惜命命【美华】【WAV+CUE】
- 晓雅《分享》DTS-WAV
- 黑鸭子2008-飞歌[首版][WAV+CUE]
- 黄乙玲1989-水泼落地难收回[日本天龙版][WAV+CUE]
- 周深《反深代词》[先听版][FLAC/分轨][310.97MB]
- 姜育恒1984《什么时候·串起又散落》台湾复刻版[WAV+CUE][1G]
- 那英《如今》引进版[WAV+CUE][1G]
- 蔡幸娟.1991-真的让我爱你吗【飞碟】【WAV+CUE】
- 群星.2024-好团圆电视剧原声带【TME】【FLAC分轨】