改名

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]);
+		$waite = new WaitGroup();
-		$this->waite(SConfig::get('kafka')[0]);
+		foreach ($kafkaServers as $kafkaServer) {
+			$waite->add();
+			go(function () use ($kafkaServer, $waite) {
+				defer(function () use ($waite) {
+					$waite->done();
+				});
+				$this->waite($kafkaServer);
+			});
+		}
+		$waite->wait();
 	}
 
 

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);
 	}