This commit is contained in:
2021-06-03 13:57:29 +08:00
parent 04ef4473cc
commit 894feb3506
+233 -220
View File
@@ -18,278 +18,291 @@ use Swoole\Timer;
abstract class Crontab extends BaseObject abstract class Crontab extends BaseObject
{ {
const WAIT_END = 'crontab:wait:execute'; const WAIT_END = 'crontab:wait:execute';
private string $name = ''; private string $name = '';
private mixed $params = []; private mixed $params = [];
private int $tickTime = 1; private int $tickTime = 1;
private bool $isLoop = false; private bool $isLoop = false;
private int $timerId = -1; private int $timerId = -1;
private int $max_execute_number = -1; private int $max_execute_number = -1;
private int $execute_number = 0; private int $execute_number = 0;
/** /**
* @return $this * @return $this
*/ */
public function increment(): static public function increment(): static
{ {
$this->execute_number += 1; $this->execute_number += 1;
return $this; return $this;
} }
/** /**
* @return string * @return string
*/ */
#[Pure] public function getName(): string #[Pure] public function getName(): string
{ {
return md5($this->name); return md5($this->name);
} }
/** /**
* @return mixed * @return mixed
*/ */
public function getParams(): mixed public function getParams(): mixed
{ {
return $this->params; return $this->params;
} }
/** /**
* @return int * @return int
*/ */
public function getTickTime(): int public function getTickTime(): int
{ {
return $this->tickTime; return $this->tickTime;
} }
/** /**
* @return bool * @return bool
*/ */
public function isLoop(): bool public function isLoop(): bool
{ {
return $this->isLoop; return $this->isLoop;
} }
/** /**
* @return int * @return int
*/ */
public function getMaxExecuteNumber(): int public function getMaxExecuteNumber(): int
{ {
return $this->max_execute_number; return $this->max_execute_number;
} }
/** /**
* @return int * @return int
*/ */
public function getExecuteNumber(): int public function getExecuteNumber(): int
{ {
return $this->execute_number; return $this->execute_number;
} }
/** /**
* @param string $name * @param string $name
*/ */
public function setName(string $name): void public function setName(string $name): void
{ {
$this->name = $name; $this->name = $name;
} }
public function setParams(): void public function setParams(): void
{ {
$this->params = func_get_args(); $this->params = func_get_args();
} }
/** /**
* @param int $tickTime * @param int $tickTime
*/ */
public function setTickTime(int $tickTime): void public function setTickTime(int $tickTime): void
{ {
$this->tickTime = $tickTime; $this->tickTime = $tickTime;
} }
/** /**
* @param bool $isLoop * @param bool $isLoop
*/ */
public function setIsLoop(bool $isLoop): void public function setIsLoop(bool $isLoop): void
{ {
$this->isLoop = $isLoop; $this->isLoop = $isLoop;
} }
/** /**
* @param int $max_execute_number * @param int $max_execute_number
*/ */
public function setMaxExecuteNumber(int $max_execute_number): void public function setMaxExecuteNumber(int $max_execute_number): void
{ {
$this->max_execute_number = $max_execute_number; $this->max_execute_number = $max_execute_number;
} }
/** /**
* @param int $execute_number * @param int $execute_number
*/ */
public function setExecuteNumber(int $execute_number): void public function setExecuteNumber(int $execute_number): void
{ {
$this->execute_number = $execute_number; $this->execute_number = $execute_number;
} }
/** /**
* @return int * @return int
*/ */
public function getTimerId(): int public function getTimerId(): int
{ {
return $this->timerId; return $this->timerId;
} }
/** /**
* @param int $timerId * @param int $timerId
*/ */
public function setTimerId(int $timerId): void public function setTimerId(int $timerId): void
{ {
$this->timerId = $timerId; $this->timerId = $timerId;
} }
/** /**
* *
* @throws Exception * @throws Exception
*/ */
public function clearTimer() public function clearTimer()
{ {
$this->warning('crontab timer clear.'); $this->warning('crontab timer clear.');
if (Timer::exists($this->timerId)) { if (Timer::exists($this->timerId)) {
Timer::clear($this->timerId); Timer::clear($this->timerId);
} }
} }
abstract public function process(): mixed; abstract public function process(): mixed;
abstract public function max_execute(): mixed; abstract public function max_execute(): mixed;
abstract public function isStop(): bool; abstract public function isStop(): bool;
/** /**
* @throws Exception * @param $name
*/ * @return mixed
public function execute(): void */
{ public function __get($name): mixed
defer(fn() => $this->afterExecute()); {
try { if (!isset($this->params[$name])) {
$redis = Snowflake::app()->getRedis(); return null;
}
$name_md5 = $this->getName(); return $this->params[$name];
}
$redis->hSet(self::WAIT_END, $name_md5, static::getSerialize($this));
$params = call_user_func([$this, 'process'], ...$this->params);
$redis->hDel(self::WAIT_END, $name_md5);
if ($params === null) {
return;
}
write(Json::encode(['name' => $this->name, 'response' => serialize($params)]), 'crontab');
} catch (\Throwable $throwable) {
logger()->addError($throwable, 'throwable');
}
}
public function afterExecute() /**
{ * @throws Exception
if ($this->isRecover() !== 999) { */
return; public function execute(): void
} {
$redis = Snowflake::app()->getRedis(); defer(fn() => $this->afterExecute());
try {
$redis = Snowflake::app()->getRedis();
$name = $this->getName(); $name_md5 = $this->getName();
if (!$redis->exists('stop:crontab:' . $name)) {
$redis->set('crontab:' . $name, swoole_serialize($this)); $redis->hSet(self::WAIT_END, $name_md5, static::getSerialize($this));
$tickTime = time() + $this->getTickTime();
$redis->zAdd(Producer::CRONTAB_KEY, $tickTime, $name); $params = call_user_func([$this, 'process'], ...$this->params);
} else { $redis->hDel(self::WAIT_END, $name_md5);
$redis->del('crontab:' . $name); if ($params === null) {
$redis->del('stop:crontab:' . $name); return;
} }
} write(Json::encode(['name' => $this->name, 'response' => serialize($params)]), 'crontab');
} catch (\Throwable $throwable) {
logger()->addError($throwable, 'throwable');
}
}
/** public function afterExecute()
* @return bool|int {
* @throws Exception if ($this->isRecover() !== 999) {
*/ return;
public function isRecover(): bool|int }
{ $redis = Snowflake::app()->getRedis();
try {
$redis = Snowflake::app()->getRedis(); $name = $this->getName();
if ($redis->exists('stop:crontab:' . $this->getName())) { if (!$redis->exists('stop:crontab:' . $name)) {
$redis->del('stop:crontab:' . $this->getName()); $redis->set('crontab:' . $name, swoole_serialize($this));
return true; $tickTime = time() + $this->getTickTime();
} $redis->zAdd(Producer::CRONTAB_KEY, $tickTime, $name);
if ($this->isExit()) { } else {
return $redis->del('crontab:' . $this->getName()); $redis->del('crontab:' . $name);
} $redis->del('stop:crontab:' . $name);
if ($this->isMaxExecute()) { }
call_user_func([$this, 'max_execute'], ...$this->getParams()); }
return $redis->del('crontab:' . $this->getName());
} else {
return 999;
}
} catch (\Throwable $throwable) {
return logger()->addError($throwable, 'throwable');
}
}
/** /**
* @param $class * @return bool|int
* @return string * @throws Exception
*/ */
public static function getSerialize($class): string public function isRecover(): bool|int
{ {
return serialize($class); try {
} $redis = Snowflake::app()->getRedis();
if ($redis->exists('stop:crontab:' . $this->getName())) {
$redis->del('stop:crontab:' . $this->getName());
return true;
}
if ($this->isExit()) {
return $redis->del('crontab:' . $this->getName());
}
if ($this->isMaxExecute()) {
call_user_func([$this, 'max_execute'], ...$this->getParams());
return $redis->del('crontab:' . $this->getName());
} else {
return 999;
}
} catch (\Throwable $throwable) {
return logger()->addError($throwable, 'throwable');
}
}
/** /**
* @return bool * @param $class
*/ * @return string
private function isExit(): bool */
{ public static function getSerialize($class): string
if ($this->isStop()) { {
return true; return serialize($class);
} }
if (!$this->isLoop) {
return true;
}
return false;
}
/** /**
* @return bool * @return bool
*/ */
private function isMaxExecute(): bool private function isExit(): bool
{ {
if ($this->max_execute_number !== -1) { if ($this->isStop()) {
return $this->execute_number >= $this->max_execute_number; return true;
} }
return false; if (!$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;
}
} }