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

变更

Browse Source
This commit is contained in:
as2252258
2022-07-10 01:19:09 +08:00
parent bd3eb54a05
commit 86798409fd
1 changed files with 131 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files
DatabasesProviders.php
+128 -109
View File
@@ -27,124 +27,143 @@ class DatabasesProviders extends Providers
{ {
/** /**
* @var EventProvider * @var EventProvider
*/ */
#[Inject(EventProvider::class)] #[Inject(EventProvider::class)]
public EventProvider $provider; public EventProvider $provider;
/** /**
* @param LocalService $application * @var PoolConnection
* @return void */
* @throws ConfigException #[Inject(PoolConnection::class)]
* @throws Exception public PoolConnection $connection;
*/
public function onImport(LocalService $application): void
{
$databases = Config::get('databases.connections', []);
if (empty($databases)) {
return;
}
$this->provider->on(OnWorkerStart::class, [$this, 'check']);
$this->provider->on(OnTaskerStart::class, [$this, 'check']);
$this->provider->on(OnWorkerExit::class, [$this, 'exit'], 9999);
foreach ($databases as $key => $database) {
$application->set($key, $this->_settings($database));
}
}
/** /**
* @param OnWorkerExit $exit * @param LocalService $application
* @return void * @return void
* @throws ConfigException * @throws ConfigException
* @throws Exception * @throws Exception
*/ */
public function exit(OnWorkerExit $exit): void public function onImport(LocalService $application): void
{ {
Timer::clearAll(); $databases = Config::get('databases.connections', []);
$databases = Config::get('databases.connections', []); if (empty($databases)) {
if (!empty($databases)) { return;
$connection = Kiri::getDi()->get(PoolConnection::class); }
foreach ($databases as $database) { $this->provider->on(OnWorkerStart::class, [$this, 'check']);
$connection->disconnect($database['cds']); $this->provider->on(OnTaskerStart::class, [$this, 'check']);
} $this->provider->on(OnWorkerExit::class, [$this, 'exit'], 9999);
} foreach ($databases as $key => $database) {
} $application->set($key, $this->_settings($database));
}
/** }
* @param $name
* @return Connection
* @throws Exception
*/
public function get($name): Connection
{
return Kiri::service()->get($name);
}
/** /**
* @param OnTaskerStart|OnWorkerStart $start * @param OnWorkerExit $exit
* @return void * @return void
*/ * @throws ConfigException
public function check(OnTaskerStart|OnWorkerStart $start): void * @throws Exception
{ */
Timer::tick(50000, function () use ($start) { public function exit(OnWorkerExit $exit): void
$databases = Config::get('databases.connections', []); {
$valid = 0; Timer::clearAll();
$count = 0; $databases = Config::get('databases.connections', []);
$logger = Kiri::getDi()->get(LoggerInterface::class); if (!empty($databases)) {
if (!empty($databases)) { foreach ($databases as $database) {
$connection = Kiri::getDi()->get(PoolConnection::class); $this->connection->disconnect($database['cds']);
foreach ($databases as $database) { }
try { }
[$total, $success] = $connection->check($database['cds']); }
$count += $total; /**
$valid += $success; * @param $name
* @return Connection
if (isset($database['slaveConfig']) && isset($database['slaveConfig']['cds'])) { * @throws Exception
if ($database['slaveConfig']['cds'] != $database['cds']) { */
[$total, $success] = $connection->check($database['slaveConfig']['cds']); public function get($name): Connection
{
$count += $total; return Kiri::service()->get($name);
$valid += $success; }
}
}
} catch (\Throwable $throwable) {
$logger->error($throwable->getMessage());
}
}
}
$logger->alert(sprintf('Worker %d db client has %d, valid %d', $start->workerId, $count, $valid));
});
}
/** /**
* @param $database * @param OnTaskerStart|OnWorkerStart $start
* @return array * @return void
*/ */
private function _settings($database): array public function check(OnTaskerStart|OnWorkerStart $start): void
{ {
$clientPool = $database['pool'] ?? ['min' => 1, 'max' => 5, 'tick' => 60]; Timer::tick(50000, static function () use ($start) {
return [ $valid = $count = 0;
'id' => $database['id'], $logger = Kiri::getDi()->get(LoggerInterface::class);
'cds' => $database['cds'], $databases = Config::get('databases.connections', []);
'class' => Connection::class, if (!empty($databases)) {
'username' => $database['username'], DatabasesProviders::each($databases, $logger, $count, $valid);
'password' => $database['password'], }
'tablePrefix' => $database['tablePrefix'], $const = 'Worker %d db client has %d, valid %d';
'database' => $database['database'], $logger->alert(sprintf($const, $start->workerId, $count, $valid));
'connect_timeout' => $database['connect_timeout'] ?? 30, });
'read_timeout' => $database['read_timeout'] ?? 10, }
'pool' => $clientPool,
'attributes' => $database['attributes'] ?? [],
'charset' => $database['charset'] ?? 'utf8mb4', /**
'slaveConfig' => $database['slaveConfig'] * @param $databases
]; * @param LoggerInterface $logger
} * @param $count
* @param $valid
* @return void
*/
public static function each($databases, LoggerInterface $logger, &$count, &$valid): void
{
$connection = Kiri::getDi()->get(PoolConnection::class);
foreach ($databases as $database) {
try {
[$total, $success] = $connection->check($database['cds']);
$count += $total;
$valid += $success;
if (isset($database['slaveConfig']) && isset($database['slaveConfig']['cds'])) {
if ($database['slaveConfig']['cds'] != $database['cds']) {
[$total, $success] = $connection->check($database['slaveConfig']['cds']);
$count += $total;
$valid += $success;
}
}
} catch (\Throwable $throwable) {
$logger->error($throwable->getMessage());
}
}
}
/**
* @param $database
* @return array
*/
private function _settings($database): array
{
$clientPool = $database['pool'] ?? ['min' => 1, 'max' => 5, 'tick' => 60];
return [
'id' => $database['id'],
'cds' => $database['cds'],
'class' => Connection::class,
'username' => $database['username'],
'password' => $database['password'],
'tablePrefix' => $database['tablePrefix'],
'database' => $database['database'],
'connect_timeout' => $database['connect_timeout'] ?? 30,
'read_timeout' => $database['read_timeout'] ?? 10,
'pool' => $clientPool,
'attributes' => $database['attributes'] ?? [],
'charset' => $database['charset'] ?? 'utf8mb4',
'slaveConfig' => $database['slaveConfig']
];
}
} }