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

modify

Browse Source
This commit is contained in:
as2252258@163.com
2021-05-04 04:10:39 +08:00
parent 72d91575a6
commit 89ea99f6ea
2 changed files with 222 additions and 232 deletions
Download Patch File Download Diff File Expand all files Collapse all files
HttpServer/Events/OnPipeMessage.php
+1 -19
View File
@@ -48,25 +48,7 @@ class OnPipeMessage extends Callback
if (empty($message['handler'] ?? null)) {
throw new Exception('unknown handler');
}
/** @var Crontab $handler */
$handler = $message['handler'];
defer(function () use ($handler) {
if ($handler->isRecover() !== 999) {
return;
}
$redis = Snowflake::app()->getRedis();
$name = $handler->getName();
if (!$redis->exists('stop:crontab:' . $name)) {
$redis->set('crontab:' . $name, swoole_serialize($handler));
$tickTime = time() + $handler->getTickTime();
$redis->zAdd(Producer::CRONTAB_KEY, $tickTime, $name);
} else {
$redis->del('crontab:' . $name);
$redis->del('stop:crontab:' . $name);
}
});
$handler->increment()->execute();
$message['handler']->increment()->execute();
return 'success';
}
System/Crontab/Crontab.php
+36 -28
View File
@@ -189,10 +189,42 @@ abstract class Crontab extends BaseObject
*/
public function execute(): void
{
defer(function () {
$this->isRecover();
});
$this->run();
defer(fn() => $this->afterExecute());
try {
$redis = Snowflake::app()->getRedis();
$name_md5 = $this->getName();
$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()
{
if ($this->isRecover() !== 999) {
return;
}
$redis = Snowflake::app()->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);
}
}
@@ -233,30 +265,6 @@ abstract class Crontab extends BaseObject
}
/**
* @throws Exception
*/
private function run()
{
try {
$redis = Snowflake::app()->getRedis();
$name_md5 = $this->getName();
$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');
}
}
/**
* @return bool
*/