as2252258/kiri-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

改名

Browse Source
This commit is contained in:
Mr·x
2020-11-10 17:52:30 +08:00
parent 04ac1877ae
commit 6cb8172b56
13 changed files with 1 additions and 543 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Queue/Abstracts/AbstractsQueue.php
-17
View File
@@ -1,17 +0,0 @@
<?php
declare(strict_types=1);
namespace Queue\Abstracts;
use Snowflake\Process\Process;
/**
* Class AbstractsQueue
* @package Queue\Abstracts
*/
abstract class AbstractsQueue extends Process
{
}
Queue/Abstracts/Queue.php
-75
View File
@@ -1,75 +0,0 @@
<?php
declare(strict_types=1);
namespace Queue\Abstracts;
use Exception;
use Queue\Consumer;
use Snowflake\Abstracts\Component;
use Snowflake\Exception\ComponentException;
use Snowflake\Snowflake;
/**
* Class Queue
* @package Queue\Abstracts
*/
abstract class Queue extends Component implements Relyon
{
/**
* @param string $key
* @param Consumer $consumer
* @param int $score
* @return false|int
* @throws ComponentException
* @throws Exception
*/
protected function push(string $key, Consumer $consumer, $score = 0)
{
$redis = Snowflake::app()->getRedis();
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();
}
}
/**
* @param $key
* @param string $consumer
* @return false|int
* @throws ComponentException
* @throws Exception
*/
protected function pop($key, string $consumer)
{
$redis = Snowflake::app()->getRedis();
try {
if (!$redis->lock($hash = md5($consumer))) {
return false;
}
$isExists = $redis->zRevRank($key, $consumer);
if ($isExists === null) {
return $redis->unlink($hash);
}
$redis->zRem($key, $consumer);
return $redis->unlink($hash);
} finally {
$redis->release();
}
}
}
Queue/Abstracts/Relyon.php
-33
View File
@@ -1,33 +0,0 @@
<?php
declare(strict_types=1);
namespace Queue\Abstracts;
use Queue\Consumer;
interface Relyon
{
// /**
// * @param string $key
// * @param Consumer $consumer
// * @param int $score
// * @return mixed
// *
// * 追加
// */
// public function push(string $key, Consumer $consumer, $score = 0);
//
//
// /**
// * @param string $key
// * @param Consumer $consumer
// * @return mixed
// * 消费
// */
// public function pop(string $key, Consumer $consumer);
//
}
Queue/Complete.php
-46
View File
@@ -1,46 +0,0 @@
<?php
declare(strict_types=1);
namespace Queue;
use Queue\Abstracts\Relyon;
use Snowflake\Exception\ComponentException;
use Snowflake\Snowflake;
/**
* Class Complete
* @package Queue
*/
class Complete extends \Queue\Abstracts\Queue
{
const QUEUE_COMPLETE = 'queue:complete:lists';
/**
* @param Consumer $consumer
* @param int $score
* @return false|int
* @throws ComponentException
*/
public function add(Consumer $consumer, $score = 0)
{
return $this->push(self::QUEUE_COMPLETE, $consumer, $score);
}
/**
* @param $consumer
* @return false|int
* @throws ComponentException
*/
public function del(string $consumer)
{
return $this->pop(self::QUEUE_COMPLETE, $consumer);
}
}
Queue/Consumer.php
-25
View File
@@ -1,25 +0,0 @@
<?php
declare(strict_types=1);
namespace Queue;
interface Consumer
{
public function __construct(array $params);
public function onWaiting();
public function onRunning();
public function onComplete();
}
Queue/Queue.php
-164
View File
@@ -1,164 +0,0 @@
<?php
declare(strict_types=1);
namespace Queue;
use Exception;
use ReflectionException;
use Snowflake\Exception\ComponentException;
use Snowflake\Exception\ConfigException;
use Snowflake\Exception\NotFindClassException;
use Snowflake\Snowflake;
use Swoole\Coroutine;
use Swoole\Process;
use Swoole\Timer;
/**
* Class Queue
* @package Queue
*/
class Queue extends \Snowflake\Process\Process
{
/** @var Waiting */
private $waiting;
/** @var Complete */
private $complete;
/** @var Running */
private $running;
private bool $shutdown = false;
/**
* Queue constructor.
* @param $application
* @param $name
* @throws ReflectionException
* @throws NotFindClassException
* @throws Exception
*/
public function __construct($application, $name)
{
parent::__construct($application, $name);
$this->waiting = Snowflake::createObject(Waiting::class);
$this->running = Snowflake::createObject(Running::class);
$this->complete = Snowflake::createObject(Complete::class);
}
/**
* @param Process $process
*/
public function onHandler(Process $process)
{
Timer::tick(50, function () {
$redis = Snowflake::app()->getRedis();
try {
$params = $redis->zRevRange(Waiting::QUEUE_WAITING, 0, 20);
if (empty($params)) {
return;
}
Coroutine::create(function () use ($params) {
$this->scheduler($params);
});
} catch (\Throwable $exception) {
$this->application->error($exception->getMessage());
} finally {
$redis->release();
}
});
}
/**
* @param Consumer $consumer
* @param int $score
* @throws ComponentException
*/
public function delivery(Consumer $consumer, $score = 0)
{
try {
$consumer->onWaiting();
} catch (\Throwable $exception) {
$this->application->error($exception->getMessage());
} finally {
$this->waiting->add($consumer, $score);
}
}
/**
* @param array $data
* @throws Exception
*/
private function scheduler(array $data)
{
$redis = Snowflake::app()->getRedis();
foreach ($data as $datum) {
$this->runner($datum);
}
$redis->release();
if ($this->shutdown === true) {
Snowflake::shutdown($this);
}
}
/**
* @param $class
* @return mixed|void
* @throws ComponentException
*/
private function runner(string $class)
{
try {
$rely_on = unserialize($class);
$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');
} finally {
$this->running->del($class);
if (isset($rely_on) && $rely_on instanceof Consumer) {
$rely_on->onComplete();
$this->complete->add($rely_on);
}
}
}
/**
* @return Waiting
*/
public function getWaiting(): Waiting
{
return $this->waiting;
}
/**
* @return Complete
*/
public function getComplete(): Complete
{
return $this->complete;
}
/**
* @return Running
*/
public function getRunning(): Running
{
return $this->running;
}
}
Queue/QueueProviders.php
-32
View File
@@ -1,32 +0,0 @@
<?php
declare(strict_types=1);
namespace Queue;
use Exception;
use HttpServer\Server;
use Snowflake\Abstracts\Providers;
use Snowflake\Application;
/**
* Class QueueProviders
* @package Queue
*/
class QueueProviders extends Providers
{
/**
* @param Application $application
* @throws Exception
*/
public function onImport(Application $application)
{
/** @var Server $server */
$server = $application->get('server');
$server->addProcess('queue', Queue::class);
}
}
Queue/Running.php
-48
View File
@@ -1,48 +0,0 @@
<?php
declare(strict_types=1);
namespace Queue;
use Exception;
use Queue\Abstracts\AbstractsQueue;
use Queue\Abstracts\Relyon;
use Snowflake\Exception\ComponentException;
use Snowflake\Snowflake;
/**
* Class Running
* @package Queue
*/
class Running extends \Queue\Abstracts\Queue
{
const QUEUE_RUNNING = 'queue:runner:lists';
/**
* @param Consumer $consumer
* @param int $score
* @return false|int
* @throws ComponentException
* @throws Exception
*/
public function add(Consumer $consumer, $score = 0)
{
return $this->push(self::QUEUE_RUNNING, $consumer, $score);
}
/**
* @param $consumer
* @return false|int
* @throws ComponentException
*/
public function del(string $consumer)
{
return $this->pop(self::QUEUE_RUNNING, $consumer);
}
}
Queue/TestQueue.php
-49
View File
@@ -1,49 +0,0 @@
<?php
declare(strict_types=1);
namespace Queue;
use Snowflake\Snowflake;
/**
* Class TestQueue
* @package Queue
*/
class TestQueue implements Consumer
{
public array $params;
/**
* TestQueue constructor.
* @param array $params
*/
public function __construct(array $params)
{
$this->params = $params;
$this->onWaiting();
}
/**
*
*/
public function onWaiting()
{
}
public function onRunning()
{
// TODO: Implement onRunning() method.
}
public function onComplete()
{
// TODO: Implement onComplete() method.
}
}
Queue/Waiting.php
-46
View File
@@ -1,46 +0,0 @@
<?php
declare(strict_types=1);
namespace Queue;
use Exception;
use Queue\Abstracts\Relyon;
use Snowflake\Exception\ComponentException;
use Snowflake\Snowflake;
/**
* Class Waiting
* @package Queue
*/
class Waiting extends \Queue\Abstracts\Queue
{
const QUEUE_WAITING = 'queue:waiting:lists';
/**
* @param Consumer $consumer
* @param int $score
* @return false|int
* @throws ComponentException
* @throws Exception
*/
public function add(Consumer $consumer, $score = 0)
{
return $this->push(self::QUEUE_WAITING, $consumer, $score);
}
/**
* @param $consumer
* @return false|int
* @throws ComponentException
*/
public function del(string $consumer)
{
return $this->pop(self::QUEUE_WAITING, $consumer);
}
}
System/Application.php
-4
View File
@@ -15,7 +15,6 @@ use Console\ConsoleProviders;
use Database\DatabasesProviders;
use Exception;
use HttpServer\ServerProviders;
use Queue\QueueProviders;
use Snowflake\Abstracts\BaseApplication;
use Snowflake\Abstracts\Config;
use Snowflake\Abstracts\Input;
@@ -50,9 +49,6 @@ class Application extends BaseApplication
$this->import(ConsoleProviders::class);
$this->import(DatabasesProviders::class);
$this->import(ServerProviders::class);
if (Config::get('queue.enable', false, false)) {
$this->import(QueueProviders::class);
}
}
System/Cache/Redis.php
-2
View File
@@ -14,9 +14,7 @@ use Snowflake\Abstracts\Component;
use Snowflake\Abstracts\Config;
use Snowflake\Event;
use Snowflake\Exception\ConfigException;
use Snowflake\Exception\RedisConnectException;
use Snowflake\Snowflake;
use Swoole\Coroutine;
/**
* Class Redis
composer.json
+1 -2
View File
@@ -9,7 +9,7 @@
],
"license": "MIT",
"require": {
"php": ">=7.4",
"php": ">=8.0",
"swoole/ide-helper": "@dev",
"ext-json": "*",
"phpmailer/phpmailer": "^6.1",
@@ -39,7 +39,6 @@
"validator\\": "Validator/",
"Console\\": "Console/",
"Database\\": "Database/",
"Queue\\": "Queue/",
"Gii\\": "Gii/",
"Kafka\\": "Kafka/"
},