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-04-20 18:38:41 +08:00
parent 819cdd835e
commit c2798e0fac
1 changed files with 58 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
System/Crontab/Zookeeper.php
+58 -54
View File
@@ -5,6 +5,7 @@ namespace Snowflake\Crontab;
use Exception; use Exception;
use Snowflake\Abstracts\Config;
use Snowflake\Cache\Redis; use Snowflake\Cache\Redis;
use Snowflake\Process\Process; use Snowflake\Process\Process;
use Snowflake\Snowflake; use Snowflake\Snowflake;
@@ -18,69 +19,72 @@ use Throwable;
class Zookeeper extends Process class Zookeeper extends Process
{ {
/** /**
* @param \Swoole\Process $process * @param \Swoole\Process $process
* @throws Exception * @throws Exception
*/ */
public function onHandler(\Swoole\Process $process): void public function onHandler(\Swoole\Process $process): void
{ {
/** @var Producer $crontab */ /** @var Producer $crontab */
$crontab = Snowflake::app()->get('crontab'); $crontab = Snowflake::app()->get('crontab');
$crontab->clearAll(); $crontab->clearAll();
if (Snowflake::getPlatform()->isLinux()) { if (Snowflake::getPlatform()->isLinux()) {
name($this->pid, 'Crontab zookeeper.'); name($this->pid, 'Crontab zookeeper.');
} }
$server = Snowflake::app()->getSwoole();
$setting = $server->setting['worker_num'] + $server->setting['task_worker_num'];
while (true) {
[$range, $redis] = $this->loadCarobTask();
foreach ($range as $value) {
$this->dispatch($server, $redis, $setting, $value);
}
$redis->release();
Coroutine::sleep(0.1); $ticker = Config::get('crontab.ticker', 100) / 1000;
}
} $server = Snowflake::app()->getSwoole();
$setting = $server->setting['worker_num'] + $server->setting['task_worker_num'];
while (true) {
[$range, $redis] = $this->loadCarobTask();
foreach ($range as $value) {
$this->dispatch($server, $redis, $setting, $value);
}
$redis->release();
Coroutine::sleep($ticker);
}
}
/** /**
* @param $server * @param $server
* @param Redis|\Redis $redis * @param Redis|\Redis $redis
* @param int $setting * @param int $setting
* @param $value * @param $value
* @throws Exception * @throws Exception
*/ */
private function dispatch($server, Redis|\Redis $redis, int $setting, $value) private function dispatch($server, Redis|\Redis $redis, int $setting, $value)
{ {
try { try {
$params['action'] = 'crontab'; $params['action'] = 'crontab';
if (empty($handler = $redis->get('crontab:' . $value))) { if (empty($handler = $redis->get('crontab:' . $value))) {
return; return;
} }
$params['handler'] = swoole_unserialize($handler); $params['handler'] = swoole_unserialize($handler);
$server->sendMessage($params, random_int(0, $setting - 1)); $server->sendMessage($params, random_int(0, $setting - 1));
} catch (Throwable $exception) { } catch (Throwable $exception) {
logger()->addError($exception); logger()->addError($exception);
} }
} }
/** /**
* @return array * @return array
* @throws Exception * @throws Exception
*/ */
private function loadCarobTask(): array private function loadCarobTask(): array
{ {
$redis = Snowflake::app()->getRedis(); $redis = Snowflake::app()->getRedis();
$range = $redis->zRangeByScore(Producer::CRONTAB_KEY, '0', (string)time()); $range = $redis->zRangeByScore(Producer::CRONTAB_KEY, '0', (string)time());
$redis->zRem(Producer::CRONTAB_KEY, ...$range); $redis->zRem(Producer::CRONTAB_KEY, ...$range);
return [$range, $redis]; return [$range, $redis];
} }
} }