modify mysql result

This commit is contained in:
2022-04-10 03:37:58 +08:00
parent 8037bfa1b9
commit f091fd5ba0
3 changed files with 295 additions and 93 deletions
+93 -80
View File
@@ -16,6 +16,9 @@ use Kiri\Server\Events\OnWorkerError;
use Kiri\Server\Events\OnWorkerExit; use Kiri\Server\Events\OnWorkerExit;
use Kiri\Server\Events\OnWorkerStart; use Kiri\Server\Events\OnWorkerStart;
use Kiri\Server\Events\OnWorkerStop; use Kiri\Server\Events\OnWorkerStop;
use Kiri\Message\Waite;
use Psr\Container\ContainerExceptionInterface;
use Psr\Container\NotFoundExceptionInterface;
use Swoole\Server; use Swoole\Server;
use Swoole\Timer; use Swoole\Timer;
@@ -28,100 +31,110 @@ class OnServerWorker extends \Kiri\Server\Abstracts\Server
{ {
public Router $collector; public Router $router;
/** /**
* @return void * @return void
*/ */
public function init() public function init()
{ {
// $this->collector = Kiri::getDi()->get(Router::class); $this->router = Kiri::getDi()->get(Router::class);
} }
/** /**
* @param Server $server * @param Server $server
* @param int $workerId * @param int $workerId
* @throws Exception * @return void
*/ * @throws Kiri\Exception\ConfigException
public function onWorkerStart(Server $server, int $workerId) * @throws ContainerExceptionInterface
{ * @throws NotFoundExceptionInterface
$dispatch = \Kiri::getDi()->get(EventDispatch::class); * @throws Exception
$dispatch->dispatch(new OnBeforeWorkerStart($workerId)); */
set_env('environmental_workerId', $workerId); public function onWorkerStart(Server $server, int $workerId)
if ($workerId < $server->setting['worker_num']) { {
set_env('environmental', Kiri::WORKER); $dispatch = \Kiri::getDi()->get(EventDispatch::class);
$this->setProcessName(sprintf('Worker Process[%d].%d', $server->worker_pid, $workerId)); $dispatch->dispatch(new OnBeforeWorkerStart($workerId));
$dispatch->dispatch(new OnWorkerStart($server, $workerId)); set_env('environmental_workerId', $workerId);
} else { if ($workerId < $server->setting['worker_num']) {
set_env('environmental', Kiri::TASK); set_env('environmental', Kiri::WORKER);
$this->setProcessName(sprintf('Tasker Process[%d].%d', $server->worker_pid, $workerId)); $this->setProcessName(sprintf('Worker Process[%d].%d', $server->worker_pid, $workerId));
$dispatch->dispatch(new OnTaskStart($server, $workerId)); $dispatch->dispatch(new OnWorkerStart($server, $workerId));
}
$dispatch->dispatch(new OnAfterWorkerStart()); $this->router->scan_build_route();
} } else {
set_env('environmental', Kiri::TASK);
$this->setProcessName(sprintf('Tasker Process[%d].%d', $server->worker_pid, $workerId));
$dispatch->dispatch(new OnTaskStart($server, $workerId));
}
$dispatch->dispatch(new OnAfterWorkerStart());
}
/** /**
* @param Server $server * @param Server $server
* @param int $workerId * @param int $workerId
* @throws Exception * @throws ContainerExceptionInterface
*/ * @throws NotFoundExceptionInterface
public function onWorkerStop(Server $server, int $workerId) */
{ public function onWorkerStop(Server $server, int $workerId)
Timer::clearAll(); {
\Kiri::getDi()->get(EventDispatch::class)->dispatch(new OnWorkerStop($server, $workerId)); Timer::clearAll();
} \Kiri::getDi()->get(EventDispatch::class)->dispatch(new OnWorkerStop($server, $workerId));
}
/** /**
* @param Server $server * @param Server $server
* @param int $workerId * @param int $workerId
* @throws Exception * @throws ContainerExceptionInterface
*/ * @throws NotFoundExceptionInterface
public function onWorkerExit(Server $server, int $workerId) */
{ public function onWorkerExit(Server $server, int $workerId)
\Kiri::getDi()->get(EventDispatch::class)->dispatch(new OnWorkerExit($server, $workerId)); {
} \Kiri::getDi()->get(EventDispatch::class)->dispatch(new OnWorkerExit($server, $workerId));
}
/** /**
* @param Server $server * @param Server $server
* @param int $worker_id * @param int $worker_id
* @param int $worker_pid * @param int $worker_pid
* @param int $exit_code * @param int $exit_code
* @param int $signal * @param int $signal
* @throws Exception * @throws ContainerExceptionInterface
*/ * @throws NotFoundExceptionInterface
public function onWorkerError(Server $server, int $worker_id, int $worker_pid, int $exit_code, int $signal) * @throws Exception
{ */
\Kiri::getDi()->get(EventDispatch::class)->dispatch(new OnWorkerError($server, $worker_id, $worker_pid, $exit_code, $signal)); public function onWorkerError(Server $server, int $worker_id, int $worker_pid, int $exit_code, int $signal)
{
\Kiri::getDi()->get(EventDispatch::class)->dispatch(new OnWorkerError($server, $worker_id, $worker_pid, $exit_code, $signal));
$message = sprintf('Worker#%d::%d error stop. signal %d, exit_code %d, msg %s', $message = sprintf('Worker#%d::%d error stop. signal %d, exit_code %d, msg %s',
$worker_id, $worker_pid, $signal, $exit_code, swoole_strerror(swoole_last_error(), 9) $worker_id, $worker_pid, $signal, $exit_code, swoole_strerror(swoole_last_error(), 9)
); );
$this->logger->error($message); $this->logger->error($message);
$this->system_mail($message); $this->system_mail($message);
} }
/** /**
* @param $messageContent * @param $messageContent
* @throws Exception * @throws Exception
*/ */
protected function system_mail($messageContent) protected function system_mail($messageContent)
{ {
try { try {
$email = Config::get('email', ['enable' => false]); $email = Config::get('email', ['enable' => false]);
if (!empty($email) && ($email['enable'] ?? false) == true) { if (!empty($email) && ($email['enable'] ?? false) == true) {
Help::sendEmail($email, 'Service Error', $messageContent); Help::sendEmail($email, 'Service Error', $messageContent);
} }
} catch (\Throwable $e) { } catch (\Throwable $e) {
error($e, 'email'); error($e, 'email');
} }
} }
} }
+188
View File
@@ -0,0 +1,188 @@
<?php
namespace Kiri\Server;
use Exception;
use Kiri\Annotation\Inject;
use Kiri\Error\StdoutLoggerInterface;
use Kiri\Server\Abstracts\BaseProcess;
use Swoole\Event;
use Swoole\Process;
use Swoole\Timer;
class Inotify extends BaseProcess
{
private mixed $inotify;
private mixed $events;
private array $watchFiles = [];
public bool $isReloading = FALSE;
public array $dirs = [];
protected int $cid;
const IG_DIR = [APP_PATH . 'commands', APP_PATH . '.git', APP_PATH . '.gitee'];
#[Inject(StdoutLoggerInterface::class)]
public StdoutLoggerInterface $logger;
/**
* @param Process $process
* @return void
* @throws Exception
*/
public function process(Process $process): void
{
// TODO: Implement process() method.
set_error_handler([$this, 'error']);
set_exception_handler([$this, 'error']);
$this->dirs = [CONTROLLER_PATH, MODEL_PATH, APP_PATH . 'routes', APP_PATH . 'vendor/game-worker'];
$this->start();
}
/**
* @return void
*/
public function error(): void
{
}
/**
* @throws Exception
*/
public function start()
{
$this->inotify = inotify_init();
$this->events = IN_MODIFY | IN_DELETE | IN_CREATE | IN_MOVE;
foreach ($this->dirs as $dir) {
if (!is_dir($dir)) continue;
$this->watch($dir);
}
Event::add($this->inotify, [$this, 'check']);
Event::wait();
}
public function clear()
{
Event::del($this->inotify);
Event::exit();
}
/**
* 开始监听
* @throws Exception
*/
public function check()
{
if (!($events = inotify_read($this->inotify))) {
return;
}
if ($this->isReloading) {
return;
}
$LISTEN_TYPE = [IN_CREATE, IN_DELETE, IN_MODIFY, IN_MOVED_TO, IN_MOVED_FROM];
foreach ($events as $ev) {
if (!in_array($ev['mask'], $LISTEN_TYPE)) {
continue;
}
//非重启类型
if (str_ends_with($ev['name'], '.php')) {
if ($this->isReloading) {
break;
}
$this->isReloading = TRUE;
Timer::after(3000, fn() => $this->reload());
}
}
}
/**
* @throws Exception
*/
public function reload()
{
$swollen = \Kiri::getDi()->get(SwooleServerInterface::class);
$swollen->reload();
$this->clearWatch();
foreach ($this->dirs as $root) {
$this->watch($root);
}
$this->isReloading = FALSE;
}
/**
* @throws Exception
*/
public function clearWatch()
{
foreach ($this->watchFiles as $wd) {
@inotify_rm_watch($this->inotify, $wd);
}
$this->watchFiles = [];
}
/**
* @param $dir
* @return bool
* @throws Exception
*/
public function watch($dir): bool
{
//目录不存在
if (!is_dir($dir)) {
return $this->logger->addError("[$dir] is not a directory.");
}
//避免重复监听
if (isset($this->watchFiles[$dir])) {
return FALSE;
}
if (in_array($dir, self::IG_DIR)) {
return FALSE;
}
$wd = @inotify_add_watch($this->inotify, $dir, $this->events);
$this->watchFiles[$dir] = $wd;
$files = scandir($dir);
foreach ($files as $f) {
if ($f == '.' || $f == '..') {
continue;
}
$path = $dir . '/' . $f;
//递归目录
if (is_dir($path)) {
$this->watch($path);
} else if (!str_ends_with($f, '.php')) {
continue;
}
//检测文件类型
if (strstr($f, '.') == '.php') {
$wd = @inotify_add_watch($this->inotify, $path, $this->events);
$this->watchFiles[$path] = $wd;
}
}
return TRUE;
}
}
+14 -13
View File
@@ -39,13 +39,16 @@ class Server extends HttpService
private mixed $daemon = 0; private mixed $daemon = 0;
/** /**
* @param State $state * @param State $state
* @param ServerManager $manager * @param ServerManager $manager
* @param ContainerInterface $container * @param ContainerInterface $container
* @param array $config * @param ProcessManager $processManager
* @throws Exception * @param EventDispatch $eventDispatch
*/ * @param Router $router
* @param array $config
* @throws Exception
*/
public function __construct(public State $state, public function __construct(public State $state,
public ServerManager $manager, public ServerManager $manager,
public ContainerInterface $container, public ContainerInterface $container,
@@ -97,8 +100,7 @@ class Server extends HttpService
* @throws ConfigException * @throws ConfigException
* @throws ContainerExceptionInterface * @throws ContainerExceptionInterface
* @throws NotFoundExceptionInterface * @throws NotFoundExceptionInterface
* @throws ReflectionException * @throws Exception
* @throws Exception
*/ */
public function start(): void public function start(): void
{ {
@@ -109,14 +111,14 @@ class Server extends HttpService
$this->manager->addListener($rpcService['type'], $rpcService['host'], $rpcService['port'], $rpcService['mode'], $rpcService); $this->manager->addListener($rpcService['type'], $rpcService['host'], $rpcService['port'], $rpcService['mode'], $rpcService);
} }
$this->process[] = Inotify::class;
$processes = array_merge($this->process, Config::get('processes', [])); $processes = array_merge($this->process, Config::get('processes', []));
$this->processManager->batch($processes); $this->processManager->batch($processes);
$this->eventDispatch->dispatch(new OnServerBeforeStart()); $this->eventDispatch->dispatch(new OnServerBeforeStart());
$this->router->scan_build_route();
$this->manager->start(); $this->manager->start();
} }
@@ -126,8 +128,7 @@ class Server extends HttpService
* @throws ConfigException * @throws ConfigException
* @throws ContainerExceptionInterface * @throws ContainerExceptionInterface
* @throws NotFoundExceptionInterface * @throws NotFoundExceptionInterface
* @throws ReflectionException * @throws Exception
* @throws Exception
*/ */
public function shutdown() public function shutdown()
{ {