改名

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);
+		try {
-		if (!$redis->lock($hash = md5($serialize))) {
+			$serialize = serialize($consumer);
-			return false;
+			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))) {
+		try {
-			return false;
+			$serialize = serialize($consumer);
-		}
+			if (!$redis->lock($hash = md5($serialize))) {
-		$isExists = $redis->zRevRank($key, $serialize);
+				return false;
-		if ($isExists === null) {
+			}
+			$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);
 	}
 
 

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) {
+			try {
-				return Timer::clear($timerId);
+				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);
+		} while (true);
-			if (empty($data)) {
-				return 1;
-			}
-			return $this->scheduler($data);
-		});
 	}