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
2021-08-25 19:11:15 +08:00
parent 84703373a2
commit bd1f60623e
22 changed files with 689 additions and 975 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/Crontab/Crontab.php
-292
View File
@@ -1,292 +0,0 @@
<?php
namespace Kiri\Crontab;
use Exception;
use JetBrains\PhpStorm\Pure;
use Server\SInterface\PipeMessage;
use Kiri\Application;
use Kiri\Kiri;
use Swoole\Timer;
/**
* Class Async
* @package Kiri
* @property Application $application
*/
abstract class Crontab implements PipeMessage, CrontabInterface
{
const WAIT_END = 'crontab:wait:execute';
private string $name = '';
private mixed $params;
private int $tickTime;
private bool $isLoop;
private int $timerId = -1;
private int $max_execute_number = -1;
private int $execute_number = 0;
/**
* Crontab constructor.
* @param mixed $params
* @param false $isLoop
* @param int $tickTime
*/
public function __construct(mixed $params, bool $isLoop = false, int $tickTime = 1)
{
$this->params = $params;
$this->isLoop = $isLoop;
$this->tickTime = $tickTime;
}
/**
* @return Application
*/
#[Pure] private function getApplication(): Application
{
return Kiri::app();
}
/**
* @return $this
*/
public function increment(): static
{
$this->execute_number += 1;
return $this;
}
/**
* @return string
*/
#[Pure] public function getName(): string
{
return md5($this->name);
}
/**
* @return mixed
*/
public function getParams(): mixed
{
return $this->params;
}
/**
* @return int
*/
public function getTickTime(): int
{
return $this->tickTime;
}
/**
* @return bool
*/
public function isLoop(): bool
{
return $this->isLoop;
}
/**
* @return int
*/
public function getMaxExecuteNumber(): int
{
return $this->max_execute_number;
}
/**
* @return int
*/
public function getExecuteNumber(): int
{
return $this->execute_number;
}
/**
* @param string $name
*/
public function setName(string $name): void
{
$this->name = $name;
}
/**
* @param int $max_execute_number
*/
public function setMaxExecuteNumber(int $max_execute_number): void
{
$this->max_execute_number = $max_execute_number;
}
/**
* @param int $execute_number
*/
public function setExecuteNumber(int $execute_number): void
{
$this->execute_number = $execute_number;
}
/**
* @return int
*/
public function getTimerId(): int
{
return $this->timerId;
}
/**
*
* @throws Exception
*/
public function clearTimer()
{
$this->application->warning('crontab timer clear.');
if (Timer::exists($this->timerId)) {
Timer::clear($this->timerId);
}
}
/**
* @param $name
* @return mixed
*/
#[Pure] public function __get($name): mixed
{
if ($name === 'application') {
return $this->getApplication();
}
if (!isset($this->params[$name])) {
return null;
}
return $this->params[$name];
}
/**
* @throws Exception
*/
public function execute(): void
{
try {
defer(fn() => $this->afterExecute());
$redis = $this->application->getRedis();
$name_md5 = $this->getName();
$redis->hSet(self::WAIT_END, $name_md5, static::getSerialize($this));
call_user_func([$this, 'process']);
$this->execute_number += 1;
$redis->hDel(self::WAIT_END, $name_md5);
} catch (\Throwable $throwable) {
$this->application->addError($throwable, 'throwable');
}
}
/**
* @throws Exception
*/
public function afterExecute()
{
if ($this->isRecover() !== 999) {
return;
}
$redis = $this->application->getRedis();
$name = $this->getName();
if (!$redis->exists('stop:crontab:' . $name)) {
$redis->set('crontab:' . $name, swoole_serialize($this));
$tickTime = time() + $this->getTickTime();
$redis->zAdd(Producer::CRONTAB_KEY, $tickTime, $name);
} else {
$redis->del('crontab:' . $name);
$redis->del('stop:crontab:' . $name);
}
}
/**
* @return bool|int
* @throws Exception
*/
public function isRecover(): bool|int
{
try {
$redis = $this->application->getRedis();
$crontab_name = $this->getName();
if ($redis->exists('stop:crontab:' . $crontab_name)) {
return $redis->del('stop:crontab:' . $crontab_name);
}
if ($this->isExit()) {
return $redis->del('crontab:' . $crontab_name);
}
if (!$this->isMaxExecute()) {
return 999;
}
call_user_func([$this, 'onMaxExecute']);
return $redis->del('crontab:' . $crontab_name);
} catch (\Throwable $throwable) {
return $this->application->addError($throwable, 'throwable');
}
}
/**
* @param $class
* @return string
*/
public static function getSerialize($class): string
{
return serialize($class);
}
/**
* @return bool
*/
private function isExit(): bool
{
if ($this->isStop() || !$this->isLoop) {
return true;
}
return false;
}
/**
* @return bool
*/
private function isMaxExecute(): bool
{
if ($this->max_execute_number !== -1) {
return $this->execute_number >= $this->max_execute_number;
}
return false;
}
}
core/Crontab/CrontabInterface.php
-20
View File
@@ -1,20 +0,0 @@
<?php
namespace Kiri\Crontab;
interface CrontabInterface
{
/**
*
*/
public function onMaxExecute(): void;
/**
* @return bool
*/
public function isStop(): bool;
}
core/Crontab/CrontabProviders.php
-40
View File
@@ -1,40 +0,0 @@
<?php
namespace Kiri\Crontab;
use Exception;
use ReflectionException;
use Kiri\Abstracts\Config;
use Kiri\Abstracts\Providers;
use Kiri\Application;
use Kiri\Exception\ComponentException;
use Kiri\Exception\ConfigException;
use Kiri\Exception\NotFindClassException;
/**
* Class CrontabProviders
* @package Kiri\Crontab
*/
class CrontabProviders extends Providers
{
/**
* @param Application $application
* @throws ConfigException
* @throws Exception
*/
public function onImport(Application $application)
{
$server = $application->getServer();
$application->set('crontab', ['class' => Producer::class]);
if (Config::get('crontab.enable') !== true) {
return;
}
$server->addProcess(Zookeeper::class);
}
}
core/Crontab/DefaultCrontab.php
-40
View File
@@ -1,40 +0,0 @@
<?php
namespace Kiri\Crontab;
/**
* Class DefaultCrontab
* @package Kiri\Crontab
*/
class DefaultCrontab extends Crontab
{
/**
* @return bool
*/
public function isStop(): bool
{
return true;
}
/**
*
*/
public function process(): void
{
// TODO: Implement process() method.
}
/**
*
*/
public function onMaxExecute(): void
{
// TODO: Implement max_execute() method.
}
}
core/Crontab/Producer.php
-96
View File
@@ -1,96 +0,0 @@
<?php
namespace Kiri\Crontab;
use Exception;
use Kiri\Abstracts\Component;
use Kiri\Kiri;
/**
* Class Producer
* @package Kiri\Abstracts
*/
class Producer extends Component
{
const CRONTAB_KEY = '_application:{crontab}:system:crontab';
/**
* @param Crontab $crontab
* @throws Exception
*/
public function dispatch(Crontab $crontab)
{
$redis = Kiri::app()->getRedis();
$name = $crontab->getName();
if ($redis->exists(self::CRONTAB_KEY) && $redis->type(self::CRONTAB_KEY) !== \Redis::REDIS_ZSET) {
throw new Exception('Cache key ' . self::CRONTAB_KEY . ' types error.');
}
$redis->del('stop:crontab:' . $name);
$redis->del('crontab:' . $name);
$redis->zRem(static::CRONTAB_KEY, $name);
$redis->zAdd(self::CRONTAB_KEY, time() + $crontab->getTickTime(), $name);
$redis->set('crontab:' . $name, swoole_serialize($crontab));
}
/**
* @param string $name
* @throws Exception
*/
public function clear(string $name)
{
$redis = Kiri::app()->getRedis();
$redis->del('crontab:' . md5($name));
$redis->zRem(static::CRONTAB_KEY, md5($name));
$redis->setex('stop:crontab:' . md5($name), 120, 1);
}
/**
* @param string $name
* @return bool
* @throws Exception
*/
public function exists(string $name): bool
{
$redis = Kiri::app()->getRedis();
if ($redis->exists('crontab:' . md5($name))) {
return true;
}
if ($redis->zRank(static::CRONTAB_KEY, md5($name))) {
return true;
}
if ($redis->hExists(Crontab::WAIT_END, md5($name))) {
return true;
}
return false;
}
/**
* @throws Exception
*/
public function clearAll()
{
$redis = Kiri::app()->getRedis();
$data = $redis->zRange(self::CRONTAB_KEY, 0, -1);
foreach ($data as $datum) {
$redis->setex('stop:crontab:' . $datum, 120, 1);
$redis->del('crontab:' . $datum);
}
$redis->release();
}
}
core/Crontab/Zookeeper.php
-129
View File
@@ -1,129 +0,0 @@
<?php
namespace Kiri\Crontab;
use Exception;
use Kiri\Abstracts\Config;
use Kiri\Cache\Redis;
use Kiri\Exception\ConfigException;
use Kiri\Kiri;
use Server\Abstracts\CustomProcess;
use Server\ServerManager;
use Swoole\Process;
use Swoole\Timer;
use Throwable;
/**
* Class Zookeeper
* @package Kiri\Process
*/
class Zookeeper extends CustomProcess
{
private int $workerNum = 0;
private int $_timer = -1;
/**
* @param Process $process
* @return string
* @throws ConfigException
*/
public function getProcessName(Process $process): string
{
$name = Config::get('id', 'system') . '[' . $process->pid . ']';
if (!empty($prefix)) {
$name .= '.Crontab zookeeper';
}
return $name;
}
/**
* @param Process $process
* @throws Exception
*/
public function onHandler(Process $process): void
{
$this->_timer = Timer::tick(300, [$this, 'loop']);
}
/**
* @throws ConfigException
* @throws Exception
*/
public function loop()
{
if ($this->checkProcessIsStop()) {
$this->exit();
Timer::clear($this->_timer);
return;
}
$redis = Kiri::app()->getRedis();
defer(fn() => $redis->release());
$range = $this->loadCarobTask($redis);
foreach ($range as $value) {
$this->dispatch($redis, $value);
}
}
/**
* @param Redis|\Redis $redis
* @param $value
* @throws Exception
*/
private function dispatch(Redis|\Redis $redis, $value)
{
try {
if (empty($handler = $redis->get('crontab:' . $value))) {
return;
}
$server = di(ServerManager::class)->getServer();
$server->sendMessage(swoole_unserialize($handler), $this->getWorker());
} catch (Throwable $exception) {
logger()->addError($exception);
}
}
/**
* @return int
* @throws Exception
*/
private function getWorker(): int
{
if ($this->workerNum == 0) {
$server = di(ServerManager::class)->getServer();
$this->workerNum = $server->setting['worker_num'] + ($server->setting['task_worker_num'] ?? 0);
}
return random_int(0, $this->workerNum - 1);
}
/**
* @param Redis|\Redis $redis
* @return array
*/
private function loadCarobTask(Redis|\Redis $redis): array
{
$script = <<<SCRIPT
local _two = redis.call('zRangeByScore', KEYS[1], '0', ARGV[1])
if (table.getn(_two) > 0) then
redis.call('ZREM', KEYS[1], unpack(_two))
end
return _two
SCRIPT;
return $redis->eval($script, [Producer::CRONTAB_KEY, (string)time()], 1);
}
}
core/Events/EventDispatch.php
-37
View File
@@ -1,37 +0,0 @@
<?php
namespace Kiri\Events;
use Annotation\Inject;
use Kiri\Abstracts\BaseObject;
/**
*
*/
class EventDispatch extends BaseObject
{
#[Inject(EventProvider::class)]
public EventProvider $eventProvider;
/**
* @param object $triggerEvent
* @return object
*/
public function dispatch(object $triggerEvent): object
{
$lists = $this->eventProvider->getListenersForEvent($triggerEvent);
foreach ($lists as $listener) {
/** @var Struct $list */
$listener($triggerEvent);
if ($triggerEvent instanceof StoppableEventInterface && $triggerEvent->isPropagationStopped()) {
break;
}
}
return $triggerEvent;
}
}
core/Events/EventInterface.php
-16
View File
@@ -1,16 +0,0 @@
<?php
namespace Kiri\Events;
/**
*
*/
interface EventInterface
{
public function process(): void;
}
core/Events/EventProvider.php
-42
View File
@@ -1,42 +0,0 @@
<?php
namespace Kiri\Events;
/**
*
*/
class EventProvider implements EventProviders
{
/** @var Struct[] */
private array $_listeners = [];
/**
* @param object $event
* @return iterable
*/
public function getListenersForEvent(object $event): iterable
{
$queue = new \SplPriorityQueue();
// TODO: Implement getListenersForEvent() method.
foreach ($this->_listeners[get_class($event)] ?? [] as $listener) {
$queue->insert($listener->listener, $listener->priority);
}
return $queue;
}
/**
* @param string $event
* @param callable $handler
* @param int $zOrder
*/
public function on(string $event, callable $handler, int $zOrder = 1)
{
$this->_listeners[$event][] = new Struct($event, $handler, $zOrder);
}
}
core/Events/EventProviders.php
-22
View File
@@ -1,22 +0,0 @@
<?php
namespace Kiri\Events;
/**
*
*/
interface EventProviders
{
/**
* @param object $event
* An event for which to return the relevant listeners.
* @return iterable<callable>
* An iterable (array, iterator, or generator) of callables. Each
* callable MUST be type-compatible with $event.
*/
public function getListenersForEvent(object $event): iterable;
}
core/Events/StoppableEventInterface.php
-24
View File
@@ -1,24 +0,0 @@
<?php
namespace Kiri\Events;
/**
*
*/
interface StoppableEventInterface
{
/**
* Is propagation stopped?
*
* This will typically only be used by the Dispatcher to determine if the
* previous listener halted propagation.
*
* @return bool
* True if the Event is complete and no further listeners should be called.
* False to continue calling listeners.
*/
public function isPropagationStopped() : bool;
}
core/Events/Struct.php
-25
View File
@@ -1,25 +0,0 @@
<?php
declare(strict_types=1);
namespace Kiri\Events;
/**
*
*/
class Struct
{
public string $event;
public array|\Closure $listener;
public int $priority;
public function __construct(string $event, callable $listener, int $priority)
{
$this->event = $event;
$this->listener = $listener;
$this->priority = $priority;
}
}