This commit is contained in:
2021-08-19 16:15:28 +08:00
parent fb453f674d
commit 99b4aee754
5 changed files with 35 additions and 36 deletions
+1 -1
View File
@@ -73,7 +73,7 @@ class Kafka extends CustomProcess
$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED); $topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);
do { do {
if ($this->isExit()) { if ($this->checkProcessIsStop()) {
$this->exit(); $this->exit();
break; break;
} }
+25 -32
View File
@@ -17,8 +17,25 @@ abstract class CustomProcess implements \Server\SInterface\CustomProcess
protected bool $enableSwooleCoroutine = true; protected bool $enableSwooleCoroutine = true;
/** @var Coroutine\Channel */ protected bool $isStop = false;
protected Coroutine\Channel $channel;
/**
*
*/
public function onProcessStop(): void
{
$this->isStop = true;
}
/**
* @return bool
*/
public function checkProcessIsStop(): bool
{
return $this->isStop === true;
}
/** /**
@@ -26,22 +43,17 @@ abstract class CustomProcess implements \Server\SInterface\CustomProcess
*/ */
public function signListen(Process $process): void public function signListen(Process $process): void
{ {
$this->channel = new Coroutine\Channel(1);
$this->channel->push(1);
if (!$this->enableSwooleCoroutine) { if (!$this->enableSwooleCoroutine) {
Process::signal(SIGTERM | SIGKILL, function ($signo) Process::signal(SIGTERM | SIGKILL, function ($signo)
use ($process) { use ($process) {
putenv('processStatus=exit'); $this->onProcessStop();
$this->waiteExit($process); $this->waiteExit($process);
}); });
} else { } else {
go(function () use ($process) { go(function () use ($process) {
$data = Coroutine::waitSignal(SIGTERM | SIGKILL, -1); $data = Coroutine::waitSignal(SIGTERM | SIGKILL, -1);
if ($data) { if ($data) {
putenv('processStatus=exit'); $this->onProcessStop();
$this->waiteExit($process); $this->waiteExit($process);
} }
}); });
@@ -49,40 +61,21 @@ abstract class CustomProcess implements \Server\SInterface\CustomProcess
} }
/**
* @return string
*/
#[Pure] protected function getStatus(): string
{
return env('processStatus', 'working');
}
/**
* @return bool
*/
#[Pure] protected function isExit(): bool
{
return $this->getStatus() == 'exit';
}
/** /**
* *
*/ */
protected function exit() protected function exit(): void
{ {
$this->channel->pop(); putenv('process.status=idle');
$this->channel->close();
} }
/** /**
* @return bool * @return bool
*/ */
public function isWorking(): bool #[Pure] public function isWorking(): bool
{ {
return $this->channel->isEmpty(); return env('process.status', 'working') == 'working';
} }
+6
View File
@@ -34,4 +34,10 @@ interface CustomProcess
public function onHandler(Process $process): void; public function onHandler(Process $process): void;
/**
*
*/
public function onProcessStop(): void;
} }
+2 -2
View File
@@ -60,9 +60,9 @@ class Zookeeper extends CustomProcess
*/ */
public function loop() public function loop()
{ {
if ($this->isExit()) { if ($this->checkProcessIsStop()) {
Timer::clear($this->_timer);
$this->exit(); $this->exit();
Timer::clear($this->_timer);
return; return;
} }
$redis = Kiri::app()->getRedis(); $redis = Kiri::app()->getRedis();
+1 -1
View File
@@ -49,7 +49,7 @@ class LoggerProcess extends CustomProcess
*/ */
public function message(Process $process) public function message(Process $process)
{ {
if ($this->isExit()) { if ($this->checkProcessIsStop()) {
$this->exit(); $this->exit();
return; return;
} }