This commit is contained in:
2020-09-11 15:01:22 +08:00
parent 48e2580c8f
commit 3bbd3e7c4e
6 changed files with 27 additions and 13 deletions
+5 -6
View File
@@ -47,24 +47,23 @@ abstract class Queue extends Component implements Relyon
/**
* @param $key
* @param Consumer $consumer
* @param string $consumer
* @return false|int
* @throws ComponentException
* @throws Exception
*/
protected function pop($key, Consumer $consumer)
protected function pop($key, string $consumer)
{
$redis = Snowflake::app()->getRedis();
try {
$serialize = serialize($consumer);
if (!$redis->lock($hash = md5($serialize))) {
if (!$redis->lock($hash = md5($consumer))) {
return false;
}
$isExists = $redis->zRevRank($key, $serialize);
$isExists = $redis->zRevRank($key, $consumer);
if ($isExists === null) {
return $redis->unlink($hash);
}
$redis->zRem($key, $serialize);
$redis->zRem($key, $consumer);
return $redis->unlink($hash);
} finally {
$redis->release();
+1 -1
View File
@@ -36,7 +36,7 @@ class Complete extends \Queue\Abstracts\Queue
* @return false|int
* @throws ComponentException
*/
public function del(Consumer $consumer)
public function del(string $consumer)
{
return $this->pop(self::QUEUE_COMPLETE, $consumer);
}
+3 -4
View File
@@ -116,19 +116,18 @@ class Queue extends \Snowflake\Process\Process
*/
private function runner(string $class)
{
$logger = $this->application->getLogger();
try {
$rely_on = unserialize($class);
$this->waiting->del($rely_on);
$this->waiting->del($class);
if (!($rely_on instanceof Consumer)) {
return;
}
$this->running->add($rely_on);
$rely_on->onRunning();
} catch (\Throwable $exception) {
$logger->write($exception->getMessage(), 'queue');
logger()->write($exception->getMessage(), 'queue');
} finally {
$this->running->del($rely_on);
$this->running->del($class);
if (isset($rely_on) && $rely_on instanceof Consumer) {
$rely_on->onComplete();
$this->complete->add($rely_on);
+1 -1
View File
@@ -38,7 +38,7 @@ class Running extends \Queue\Abstracts\Queue
* @return false|int
* @throws ComponentException
*/
public function del(Consumer $consumer)
public function del(string $consumer)
{
return $this->pop(self::QUEUE_RUNNING, $consumer);
}
+1 -1
View File
@@ -37,7 +37,7 @@ class Waiting extends \Queue\Abstracts\Queue
* @return false|int
* @throws ComponentException
*/
public function del(Consumer $consumer)
public function del(string $consumer)
{
return $this->pop(self::QUEUE_WAITING, $consumer);
}
+16
View File
@@ -3,6 +3,8 @@
defined('APP_PATH') or define('APP_PATH', __DIR__ . '/../../');
use HttpServer\Http\Response;
use Snowflake\Error\Logger;
use Snowflake\Exception\ComponentException;
use Snowflake\Snowflake;
use HttpServer\Http\Context;
use Snowflake\Core\ArrayAccess;
@@ -45,6 +47,20 @@ if (!function_exists('exif_imagetype')) {
}
}
if (!function_exists('logger')) {
/**
* @return Logger
* @throws ComponentException
*/
function logger()
{
return Snowflake::app()->getLogger();
}
}
if (!function_exists('get_file_extension')) {
function get_file_extension($filename)