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
2020-10-28 11:46:43 +08:00
parent 39ad35393f
commit 8f820a3758
1 changed files with 42 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Kafka/Kafka.php
+42 -31
View File
@@ -58,9 +58,18 @@ class Kafka extends \Snowflake\Process\Process
$this->channelListener(); $this->channelListener();
$kafkaServers = SConfig::get('kafka'); $kafkaServers = SConfig::get('kafka');
$waite = new WaitGroup();
foreach ($kafkaServers as $kafkaServer) { foreach ($kafkaServers as $kafkaServer) {
$this->waite($kafkaServer); $waite->add();
go(function () use ($kafkaServer, $waite) {
defer(function () use ($waite) {
$waite->done();
});
$this->waite($kafkaServer);
});
} }
$waite->wait();
} }
@@ -69,37 +78,39 @@ class Kafka extends \Snowflake\Process\Process
*/ */
private function waite(array $kafkaServer) private function waite(array $kafkaServer)
{ {
go(function () use ($kafkaServer) { [$config, $topic, $conf] = $this->kafkaConfig($kafkaServer);
[$config, $topic, $conf] = $this->kafkaConfig($kafkaServer); $objRdKafka = new \RdKafka\Consumer($config);
$objRdKafka = new \RdKafka\Consumer($config); $topic = $objRdKafka->newTopic($kafkaServer['topic'], $topic);
$topic = $objRdKafka->newTopic($kafkaServer['topic'], $topic); $topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);
$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED); while (true) {
while (true) { $this->resolve($topic);
try { }
$message = $topic->consume(0, $conf['metadataRefreshIntervalMs'] ?? 1000); }
if (empty($message)) {
$this->application->debug('message null.');
continue; /**
} * @param ConsumerTopic $topic
var_dump($message); */
switch ($message->err) { private function resolve(ConsumerTopic $topic)
case RD_KAFKA_RESP_ERR_NO_ERROR: {
$this->channel->push([$message->topic_name, $message]); try {
break; $message = $topic->consume(0, $conf['interval'] ?? 1000);
case RD_KAFKA_RESP_ERR__PARTITION_EOF: if (empty($message)) {
$this->application->error('No more messages; will wait for more'); $this->application->debug('message null.');
break; return;
case RD_KAFKA_RESP_ERR__TIMED_OUT:
$this->application->error('Kafka Timed out');
break;
default:
throw new \Exception($message->errstr(), $message->err);
}
} catch (\Throwable $exception) {
$this->application->error($exception->getMessage());
}
} }
}); if ($message->err == RD_KAFKA_RESP_ERR_NO_ERROR) {
$this->channel->push([$message->topic_name, $message]);
} else if ($message->err == RD_KAFKA_RESP_ERR__PARTITION_EOF) {
$this->application->error('No more messages; will wait for more');
} else if ($message->err == RD_KAFKA_RESP_ERR__TIMED_OUT) {
$this->application->error('Kafka Timed out');
} else {
$this->application->error($message->errstr());
}
} catch (\Throwable $exception) {
$this->application->error($exception->getMessage());
}
} }