From 9df3b56ec0136574055f1ac3c0dbcade82c7c199 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Mr=C2=B7x?= <as2252258@163.com>
Date: Fri, 11 Sep 2020 01:25:43 +0800
Subject: [PATCH] =?UTF-8?q?=E6=94=B9=E5=90=8D?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 Queue/Abstracts/Queue.php | 40 +++++++++++++++++++++++----------------
 Queue/Queue.php           | 28 ++++++++++++++++++---------
 2 files changed, 43 insertions(+), 25 deletions(-)

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);
 	}