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-04-16 00:22:26 +08:00
parent e9243de9c1
commit de6e01b058
4 changed files with 273 additions and 275 deletions
Download Patch File Download Diff File Expand all files Collapse all files
HttpServer/Events/OnPipeMessage.php
+53 -55
View File
@@ -18,70 +18,68 @@ use Swoole\Server;
class OnPipeMessage extends Callback class OnPipeMessage extends Callback
{ {
/** /**
* @param Server $server * @param Server $server
* @param int $src_worker_id * @param int $src_worker_id
* @param $swollen_universalize * @param $swollen_universalize
* @throws Exception * @throws Exception
*/ */
public function onHandler(Server $server, int $src_worker_id, $swollen_universalize) public function onHandler(Server $server, int $src_worker_id, $swollen_universalize)
{ {
go(function () use ($server, $src_worker_id, $swollen_universalize) { go(function () use ($server, $src_worker_id, $swollen_universalize) {
match ($swollen_universalize['action'] ?? null) { match ($swollen_universalize['action'] ?? null) {
'kafka' => $this->onKafkaWorker($swollen_universalize), 'kafka' => $this->onKafkaWorker($swollen_universalize),
'crontab' => $this->onCrontabWorker($swollen_universalize), 'crontab' => $this->onCrontabWorker($swollen_universalize),
default => $this->onMessageWorker($server, $src_worker_id, $swollen_universalize) default => $this->onMessageWorker($server, $src_worker_id, $swollen_universalize)
}; };
fire(Event::SYSTEM_RESOURCE_RELEASES); fire(Event::SYSTEM_RESOURCE_RELEASES);
}); });
} }
/** /**
* @param array $message * @param array $message
* @return string * @return string
* @throws Exception * @throws Exception
*/ */
private function onCrontabWorker(array $message): string private function onCrontabWorker(array $message): string
{ {
if (!isset($message['handler'])) { if (empty($handler = $message['handler'] ?? null)) {
throw new Exception('unknown handler'); throw new Exception('unknown handler');
} }
/** @var Crontab $handler */
/** @var Crontab $crontab */ $handler->increment()->execute();
$crontab = swoole_unserialize($message['handler']); return 'success';
$crontab->increment()->execute(); }
return 'success';
}
/** /**
* @param $server * @param $server
* @param $src_worker_id * @param $src_worker_id
* @param $message * @param $message
* @return string * @return string
* @throws Exception * @throws Exception
*/ */
private function onMessageWorker($server, $src_worker_id, $message): string private function onMessageWorker($server, $src_worker_id, $message): string
{ {
fire(Event::PIPE_MESSAGE, [$server, $src_worker_id, $message]); fire(Event::PIPE_MESSAGE, [$server, $src_worker_id, $message]);
return 'success'; return 'success';
} }
/** /**
* @param array $message * @param array $message
* @return string * @return string
*/ */
private function onKafkaWorker(array $message): string private function onKafkaWorker(array $message): string
{ {
[$topic, $rdMessage] = $message['body']; [$topic, $rdMessage] = $message['body'];
call_user_func($message['handler'], new Struct($topic, $rdMessage)); call_user_func($message['handler'], new Struct($topic, $rdMessage));
return 'success'; return 'success';
} }
} }
System/Crontab/Crontab.php
+218 -217
View File
@@ -18,277 +18,278 @@ 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);
} }
} }
/** /**
* @throws Exception * @throws Exception
*/ */
private function recover() private function recover()
{ {
$redis = Snowflake::app()->getRedis(); $redis = Snowflake::app()->getRedis();
if (!$redis->exists('stop:crontab:' . $this->getName())) { if (!$redis->exists('stop:crontab:' . $this->getName())) {
$redis->del('stop:crontab:' . $this->getName()); $redis->del('crontab:' . $this->getName());
return; $redis->del('stop:crontab:' . $this->getName());
} } else {
$redis->set('crontab:' . ($name = $this->getName()), swoole_serialize($this));
$redis->set('crontab:' . ($name = $this->getName()), swoole_serialize($this)); $tickTime = time() + $this->getTickTime();
$redis->zAdd(Producer::CRONTAB_KEY, $tickTime, $name);
$tickTime = time() + $this->getTickTime(); }
}
$redis->zAdd(Producer::CRONTAB_KEY, $tickTime, $name);
}
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 * @throws Exception
*/ */
public function execute(): void public function execute(): void
{ {
\Swoole\Coroutine\go(function ($application) { \Swoole\Coroutine\go(function ($application) {
$application->isRecover($application); $application->isRecover($application);
}, $this); }, $this);
$this->run(); $this->run();
} }
/** /**
* @param $application * @param $application
* @throws Exception * @return bool|int
*/ * @throws \Exception
public function isRecover($application) */
{ public function isRecover($application): bool|int
try { {
if ($application->isExit()) { try {
return; $redis = Snowflake::app()->getRedis();
} if ($application->isExit()) {
if ($application->isMaxExecute()) { return $redis->del('crontab:' . $this->getName());
call_user_func([$application, 'max_execute'], ...$application->params); }
} else { if ($application->isMaxExecute()) {
$application->recover(); call_user_func([$application, 'max_execute'], ...$application->params);
} return $redis->del('crontab:' . $this->getName());
} catch (\Throwable $throwable) { } else {
logger()->addError($throwable, 'throwable'); return $application->recover();
} }
} } catch (\Throwable $throwable) {
return logger()->addError($throwable, 'throwable');
}
}
/** /**
* @throws Exception * @throws Exception
*/ */
private function run() private function run()
{ {
try { try {
$redis = Snowflake::app()->getRedis(); $redis = Snowflake::app()->getRedis();
$name_md5 = $this->getName(); $name_md5 = $this->getName();
$redis->hSet(self::WAIT_END, $name_md5, serialize($this)); $redis->hSet(self::WAIT_END, $name_md5, serialize($this));
$params = call_user_func([$this, 'process'], ...$this->params); $params = call_user_func([$this, 'process'], ...$this->params);
$redis->hDel(self::WAIT_END, $name_md5); $redis->hDel(self::WAIT_END, $name_md5);
if ($params === null) { if ($params === null) {
return; return;
} }
$name = date('Y-m-d.log'); $name = date('Y-m-d.log');
write(storage($name, '/log/crontab'), Json::encode([ write(storage($name, '/log/crontab'), Json::encode([
'name' => $this->name, 'name' => $this->name,
'response' => serialize($params) 'response' => serialize($params)
])); ]));
} catch (\Throwable $throwable) { } catch (\Throwable $throwable) {
logger()->addError($throwable, 'throwable'); logger()->addError($throwable, 'throwable');
} }
} }
/** /**
* @return bool * @return bool
*/ */
private function isExit(): bool private function isExit(): bool
{ {
if ($this->isStop()) { if ($this->isStop()) {
return true; return true;
} }
if (!$this->isLoop) { if (!$this->isLoop) {
return true; return true;
} }
return false; return false;
} }
/** /**
* @return bool * @return bool
*/ */
private function isMaxExecute(): bool private function isMaxExecute(): bool
{ {
if ($this->max_execute_number !== -1) { if ($this->max_execute_number !== -1) {
return $this->execute_number >= $this->max_execute_number; return $this->execute_number >= $this->max_execute_number;
} }
return false; return false;
} }
} }
System/Crontab/Producer.php
-1
View File
@@ -35,7 +35,6 @@ class Producer extends Component
} }
$tickTime = time() + $crontab->getTickTime(); $tickTime = time() + $crontab->getTickTime();
$redis->del('stop:crontab:' . $name, 120); $redis->del('stop:crontab:' . $name, 120);
$result = $redis->zAdd(self::CRONTAB_KEY, $tickTime, $name); $result = $redis->zAdd(self::CRONTAB_KEY, $tickTime, $name);
System/Crontab/Zookeeper.php
+2 -2
View File
@@ -67,8 +67,8 @@ class Zookeeper extends Process
{ {
try { try {
$params['action'] = 'crontab'; $params['action'] = 'crontab';
$params['handler'] = $redis->get('crontab:' . $value); $params['handler'] = swoole_unserialize($redis->get('crontab:' . $value));
$redis->del('crontab:' . $value);
$result = $server->sendMessage($params, $workerId = random_int(0, $setting - 1)); $result = $server->sendMessage($params, $workerId = random_int(0, $setting - 1));
var_dump('send crontab to ' . $workerId . ' ' . intval($result)); var_dump('send crontab to ' . $workerId . ' ' . intval($result));