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-28 12:08:10 +08:00
parent 66dfa660d5
commit 872e1a4ea0
1 changed files with 74 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
System/Crontab/Zookeeper.php
+74 -67
View File
@@ -5,13 +5,12 @@ namespace Snowflake\Crontab;
use Exception; use Exception;
use HttpServer\Server;
use Snowflake\Abstracts\Config; use Snowflake\Abstracts\Config;
use Snowflake\Cache\Redis; use Snowflake\Cache\Redis;
use Snowflake\Exception\ConfigException; use Snowflake\Exception\ConfigException;
use Snowflake\Process\Process; use Snowflake\Process\Process;
use Snowflake\Snowflake; use Snowflake\Snowflake;
use Swoole\Coroutine; use Swoole\Timer;
use Throwable; use Throwable;
/** /**
@@ -22,57 +21,65 @@ class Zookeeper extends Process
{ {
private int $workerNum = 0; private int $workerNum = 0;
private mixed $server; private mixed $server;
/** /**
* @return string * @return string
* @throws ConfigException * @throws ConfigException
*/ */
public function getProcessName(): string public function getProcessName(): string
{ {
$name = Config::get('id', 'system') . '[' . $this->pid . ']'; $name = Config::get('id', 'system') . '[' . $this->pid . ']';
if (!empty($prefix)) { if (!empty($prefix)) {
$name .= '.Crontab zookeeper'; $name .= '.Crontab zookeeper';
} }
return $name; return $name;
} }
/** /**
* @param \Swoole\Process $process * @param \Swoole\Process $process
* @throws Exception * @throws Exception
*/ */
public function before(\Swoole\Process $process): void public function before(\Swoole\Process $process): void
{ {
/** @var Producer $crontab */ /** @var Producer $crontab */
$crontab = Snowflake::app()->get('crontab'); $crontab = Snowflake::app()->get('crontab');
$crontab->clearAll(); $crontab->clearAll();
$this->server = $server = Snowflake::app()->getSwoole(); $this->server = $server = Snowflake::app()->getSwoole();
$this->workerNum = $server->setting['worker_num'] + $server->setting['task_worker_num']; $this->workerNum = $server->setting['worker_num'] + $server->setting['task_worker_num'];
} }
/** /**
* @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
{ {
$redis = Snowflake::app()->getRedis(); Timer::tick(80, [$this, 'loop']);
while (true) { }
$range = $this->loadCarobTask($redis);
var_dump($range);
foreach ($range as $value) { /**
$this->dispatch($redis, $value); * @throws ConfigException
} * @throws Exception
Coroutine::sleep(80); */
} public function loop()
} {
$redis = Snowflake::app()->getRedis();
defer(fn() => $redis->release());
$range = $this->loadCarobTask($redis);
var_dump($range);
foreach ($range as $value) {
$this->dispatch($redis, $value);
}
}
/** /**
@@ -80,43 +87,43 @@ class Zookeeper extends Process
* @param $value * @param $value
* @throws Exception * @throws Exception
*/ */
private function dispatch(Redis|\Redis $redis, $value) private function dispatch(Redis|\Redis $redis, $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);
$this->server->sendMessage($params, $this->getWorker()); $this->server->sendMessage($params, $this->getWorker());
} catch (Throwable $exception) { } catch (Throwable $exception) {
logger()->addError($exception); logger()->addError($exception);
} }
} }
/** /**
* @return int * @return int
* @throws \Exception * @throws \Exception
*/ */
private function getWorker(): int private function getWorker(): int
{ {
return random_int(0, $this->workerNum - 1); return random_int(0, $this->workerNum - 1);
} }
/** /**
* @param Redis|\Redis $redis * @param Redis|\Redis $redis
* @return array * @return array
*/ */
private function loadCarobTask(Redis|\Redis $redis): array private function loadCarobTask(Redis|\Redis $redis): array
{ {
$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; return $range;
} }
} }