改名

core/Crontab/Crontab.php

@@ -0,0 +1,292 @@
+<?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

@@ -0,0 +1,20 @@
+<?php
+
+namespace Kiri\Crontab;
+
+interface CrontabInterface
+{
+
+
+	/**
+	 *
+	 */
+	public function onMaxExecute(): void;
+
+
+	/**
+	 * @return bool
+	 */
+	public function isStop(): bool;
+
+}

core/Crontab/CrontabProviders.php

@@ -0,0 +1,40 @@
+<?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

@@ -0,0 +1,40 @@
+<?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

@@ -0,0 +1,96 @@
+<?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

@@ -0,0 +1,129 @@
+<?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);
+	}
+
+}