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-08-19 18:46:28 +08:00
parent 63fd1e8e13
commit 11568decb4
3 changed files with 19 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Kafka/Kafka.php
+13 -13
View File
@@ -14,6 +14,7 @@ use RdKafka\Exception;
use RdKafka\KafkaConsumer; use RdKafka\KafkaConsumer;
use RdKafka\TopicConf; use RdKafka\TopicConf;
use Server\Abstracts\CustomProcess; use Server\Abstracts\CustomProcess;
use Swoole\Coroutine;
use Swoole\Process; use Swoole\Process;
use Throwable; use Throwable;
@@ -73,8 +74,7 @@ class Kafka extends CustomProcess
$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED); $topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);
do { do {
if ($this->checkProcessIsStop()) { if ($this->isStop) {
$this->exit();
break; break;
} }
$this->resolve($topic, $conf['interval'] ?? 1000); $this->resolve($topic, $conf['interval'] ?? 1000);
@@ -94,18 +94,18 @@ class Kafka extends CustomProcess
{ {
try { try {
$message = $topic->consume(0, $interval); $message = $topic->consume(0, $interval);
if (empty($message)) { if (!empty($message)) {
return; if ($message->err == RD_KAFKA_RESP_ERR_NO_ERROR) {
} $this->handlerExecute($message->topic_name, $message);
if ($message->err == RD_KAFKA_RESP_ERR_NO_ERROR) { } else if ($message->err == RD_KAFKA_RESP_ERR__PARTITION_EOF) {
$this->handlerExecute($message->topic_name, $message); logger()->warning('No more messages; will wait for more');
} else if ($message->err == RD_KAFKA_RESP_ERR__PARTITION_EOF) { } else if ($message->err == RD_KAFKA_RESP_ERR__TIMED_OUT) {
logger()->warning('No more messages; will wait for more'); logger()->error('Kafka Timed out');
} else if ($message->err == RD_KAFKA_RESP_ERR__TIMED_OUT) { } else {
logger()->error('Kafka Timed out'); logger()->error($message->errstr());
} else { }
logger()->error($message->errstr());
} }
Coroutine::sleep(0.01);
} catch (Throwable $exception) { } catch (Throwable $exception) {
logger()->addError($exception, 'throwable'); logger()->addError($exception, 'throwable');
} }
Server/Abstracts/CustomProcess.php
+5 -3
View File
@@ -57,8 +57,9 @@ abstract class CustomProcess implements \Server\SInterface\CustomProcess
Coroutine::create(function () use ($process) { Coroutine::create(function () use ($process) {
/** @var Coroutine\Socket $message */ /** @var Coroutine\Socket $message */
$message = $process->exportSocket(); $message = $process->exportSocket();
error($message->recv()); if ($message->recv() == 0x03455343213212) {
$process->exit(0); $this->waiteExit($process);
}
}); });
Coroutine::create(function () use ($process) { Coroutine::create(function () use ($process) {
$data = Coroutine::waitSignal(SIGTERM | SIGKILL, -1); $data = Coroutine::waitSignal(SIGTERM | SIGKILL, -1);
@@ -68,7 +69,7 @@ abstract class CustomProcess implements \Server\SInterface\CustomProcess
foreach ($process as $item) { foreach ($process as $item) {
/** @var Coroutine\Socket $export */ /** @var Coroutine\Socket $export */
$export = $item->exportSocket(); $export = $item->exportSocket();
$export->send([$name => 'exit']); $export->send(0x03455343213212);
} }
} }
} }
@@ -100,6 +101,7 @@ abstract class CustomProcess implements \Server\SInterface\CustomProcess
*/ */
private function waiteExit(Process $process): void private function waiteExit(Process $process): void
{ {
$this->onProcessStop();
while ($this->isWorking()) { while ($this->isWorking()) {
$this->sleep(); $this->sleep();
} }
System/Application.php
+1 -1
View File
@@ -88,7 +88,7 @@ class Application extends BaseApplication
*/ */
public function addProcess(string $class, Process $process) public function addProcess(string $class, Process $process)
{ {
if (isset($this->_process[$class])) { if (!isset($this->_process[$class])) {
$this->_process[$class] = []; $this->_process[$class] = [];
} }
$this->_process[$class][] = $process; $this->_process[$class][] = $process;