From 3b5c588eebbbda01a1918e0a7f7da7dfea702e49 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mr=C2=B7x?= Date: Wed, 28 Oct 2020 11:52:34 +0800 Subject: [PATCH] =?UTF-8?q?=E6=94=B9=E5=90=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Kafka/Kafka.php | 14 ++++++++++++-- Kafka/KafkaProviders.php | 15 --------------- 2 files changed, 12 insertions(+), 17 deletions(-) diff --git a/Kafka/Kafka.php b/Kafka/Kafka.php index 070f29c2..c76d25ed 100644 --- a/Kafka/Kafka.php +++ b/Kafka/Kafka.php @@ -56,9 +56,19 @@ class Kafka extends \Snowflake\Process\Process public function onHandler(Process $process) { $this->channelListener(); + $kafkaServers = SConfig::get('kafka'); - var_dump(SConfig::get('kafka')[0]); - $this->waite(SConfig::get('kafka')[0]); + $waite = new WaitGroup(); + foreach ($kafkaServers as $kafkaServer) { + $waite->add(); + go(function () use ($kafkaServer, $waite) { + defer(function () use ($waite) { + $waite->done(); + }); + $this->waite($kafkaServer); + }); + } + $waite->wait(); } diff --git a/Kafka/KafkaProviders.php b/Kafka/KafkaProviders.php index c7874152..f080e012 100644 --- a/Kafka/KafkaProviders.php +++ b/Kafka/KafkaProviders.php @@ -31,21 +31,6 @@ class KafkaProviders extends Providers if (empty($kafka) || !($kafka['enable'] ?? false)) { return; } - if (!isset($kafka['topic']) || empty($kafka['topic'])) { - throw new Exception('kafka configure error.'); - } - if (!isset($kafka['version']) || empty($kafka['version'])) { - throw new Exception('kafka configure error.'); - } - if (!isset($kafka['brokers']) || empty($kafka['brokers'])) { - throw new Exception('kafka configure error.'); - } - if (!isset($kafka['groupId']) || empty($kafka['groupId'])) { - throw new Exception('kafka configure error.'); - } - if (!is_array($kafka['topics'])) { - throw new Exception('Add kafka topics must is array.'); - } $server->addProcess('kafka', Kafka::class); }