改名

Kafka/Kafka.php

@@ -41,7 +41,6 @@ class Kafka extends \Snowflake\Process\Process
 
 		$waite = new WaitGroup();
 		$kafkaServers = SConfig::get('kafka.servers');
-		var_dump($kafkaServers);
 		foreach ($kafkaServers as $kafkaServer) {
 			$waite->add();
 			go(function () use ($kafkaServer, $waite) {
@@ -60,6 +59,7 @@ class Kafka extends \Snowflake\Process\Process
 	 */
 	private function waite(array $kafkaServer)
 	{
+		try {
 			[$config, $topic, $conf] = $this->kafkaConfig($kafkaServer);
 			$objRdKafka = new Consumer($config);
 			$topic = $objRdKafka->newTopic($kafkaServer['topic'], $topic);
@@ -70,6 +70,9 @@ class Kafka extends \Snowflake\Process\Process
 			while (true) {
 				$this->resolve($topic, $conf['interval'] ?? 1000);
 			}
+		} catch (Throwable $exception) {
+			$this->application->error($exception->getMessage());
+		}
 	}
 
 
@@ -85,7 +88,6 @@ class Kafka extends \Snowflake\Process\Process
 				return;
 			}
 			if ($message->err == RD_KAFKA_RESP_ERR_NO_ERROR) {
-				var_dump($message->topic_name, $message);
 				$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');