diff --git a/Queue/Abstracts/Queue.php b/Queue/Abstracts/Queue.php index 60c46e03..e8517bad 100644 --- a/Queue/Abstracts/Queue.php +++ b/Queue/Abstracts/Queue.php @@ -29,15 +29,19 @@ abstract class Queue extends Component implements Relyon protected function push(string $key, Consumer $consumer, $score = 0) { $redis = Snowflake::app()->getRedis(); - $serialize = serialize($consumer); - if (!$redis->lock($hash = md5($serialize))) { - return false; + try { + $serialize = serialize($consumer); + if (!$redis->lock($hash = md5($serialize))) { + return false; + } + $isExists = $redis->zRevRank($key, $serialize); + if ($isExists !== null) { + $redis->zAdd($key, $score, $serialize); + } + return $redis->unlink($hash); + } finally { + $redis->release(); } - $isExists = $redis->zRevRank($key, $serialize); - if ($isExists !== null) { - $redis->zAdd($key, $score, $serialize); - } - return $redis->unlink($hash); } @@ -50,17 +54,21 @@ abstract class Queue extends Component implements Relyon */ protected function pop($key, Consumer $consumer) { - $serialize = serialize($consumer); $redis = Snowflake::app()->getRedis(); - if (!$redis->lock($hash = md5($serialize))) { - return false; - } - $isExists = $redis->zRevRank($key, $serialize); - if ($isExists === null) { + try { + $serialize = serialize($consumer); + if (!$redis->lock($hash = md5($serialize))) { + return false; + } + $isExists = $redis->zRevRank($key, $serialize); + if ($isExists === null) { + return $redis->unlink($hash); + } + $redis->zRem($key, $serialize); return $redis->unlink($hash); + } finally { + $redis->release(); } - $redis->zRem($key, $serialize); - return $redis->unlink($hash); } diff --git a/Queue/Queue.php b/Queue/Queue.php index 1de549ff..304a1646 100644 --- a/Queue/Queue.php +++ b/Queue/Queue.php @@ -7,6 +7,7 @@ namespace Queue; use Exception; use ReflectionException; use Snowflake\Exception\ComponentException; +use Snowflake\Exception\ConfigException; use Snowflake\Exception\NotFindClassException; use Snowflake\Snowflake; use Swoole\Coroutine; @@ -56,20 +57,29 @@ class Queue extends \Snowflake\Process\Process /** * @param Process $process + * @throws ComponentException + * @throws ConfigException */ public function onHandler(Process $process) { - Timer::tick(50, function ($timerId) { + do { $redis = Snowflake::app()->getRedis(); - if ($this->shutdown) { - return Timer::clear($timerId); + try { + if ($this->shutdown) { + return; + } + $data = $redis->zRevRange(Waiting::QUEUE_WAITING, 0, 20); + if (empty($data)) { + Coroutine::sleep(0.05); + } else { + $this->scheduler($data); + } + } catch (\Throwable $exception) { + $this->application->error($exception->getMessage()); + } finally { + $redis->release(); } - $data = $redis->zRevRange(Waiting::QUEUE_WAITING, 0, 20); - if (empty($data)) { - return 1; - } - return $this->scheduler($data); - }); + } while (true); }