From 8572bb21447282c46dcb8b98ba133b172314004a Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Mr=C2=B7x?= <as2252258@163.com>
Date: Wed, 28 Oct 2020 11:40:33 +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 | 79 +++++++++++++++++++++++++++++--------------------
 1 file changed, 47 insertions(+), 32 deletions(-)

diff --git a/Kafka/Kafka.php b/Kafka/Kafka.php
index 16a11436..48b5eb96 100644
--- a/Kafka/Kafka.php
+++ b/Kafka/Kafka.php
@@ -5,6 +5,7 @@ namespace Kafka;
 
 
 use RdKafka\Conf;
+use RdKafka\ConsumerTopic;
 use RdKafka\KafkaConsumer;
 use RdKafka\TopicConf;
 use Snowflake\Exception\ConfigException;
@@ -14,6 +15,7 @@ use Swoole\Coroutine\Channel;
 use Swoole\Coroutine\WaitGroup;
 use Swoole\Process;
 use Snowflake\Abstracts\Config as SConfig;
+use Swoole\Timer;
 use function Amp\stop;
 
 /**
@@ -55,33 +57,52 @@ class Kafka extends \Snowflake\Process\Process
 	public function onHandler(Process $process)
 	{
 		$this->channelListener();
-		[$config, $topic, $conf] = $this->kafkaConfig();
-		$objRdKafka = new \RdKafka\Consumer($config);
-		$topic = $objRdKafka->newTopic('test', $topic);
-		$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);
-		while (true) {
-			$message = $topic->consume(0, $conf['metadataRefreshIntervalMs'] ?? 1000);
-			if (empty($message)) {
-				$this->application->debug('message null.');
-				continue;
-			}
-			switch ($message->err) {
-				case RD_KAFKA_RESP_ERR_NO_ERROR:
-					$this->channel->push([$message->topic_name, $message]);
-					break;
-				case RD_KAFKA_RESP_ERR__PARTITION_EOF:
-					$this->application->error('No more messages; will wait for more');
-					break;
-				case RD_KAFKA_RESP_ERR__TIMED_OUT:
-					$this->application->error('Kafka Timed out');
-					break;
-				default:
-					throw new \Exception($message->errstr(), $message->err);
-			}
+
+		$kafkaServers = SConfig::get('kafka');
+		foreach ($kafkaServers as $kafkaServer) {
+			$this->waite($kafkaServer);
 		}
 	}
 
 
+	/**
+	 * @param array $kafkaServer
+	 */
+	private function waite(array $kafkaServer)
+	{
+		go(function () use ($kafkaServer) {
+			[$config, $topic, $conf] = $this->kafkaConfig($kafkaServer);
+			$objRdKafka = new \RdKafka\Consumer($config);
+			$topic = $objRdKafka->newTopic($kafkaServer['topic'], $topic);
+			$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);
+			while (true) {
+				try {
+					$message = $topic->consume(0, $conf['metadataRefreshIntervalMs'] ?? 1000);
+					if (empty($message)) {
+						$this->application->debug('message null.');
+						return;
+					}
+					switch ($message->err) {
+						case RD_KAFKA_RESP_ERR_NO_ERROR:
+							$this->channel->push([$message->topic_name, $message]);
+							break;
+						case RD_KAFKA_RESP_ERR__PARTITION_EOF:
+							$this->application->error('No more messages; will wait for more');
+							break;
+						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());
+				}
+			}
+		});
+	}
+
+
 	/**
 	 * 监听通道数据传递
 	 */
@@ -129,14 +150,12 @@ class Kafka extends \Snowflake\Process\Process
 
 
 	/**
+	 * @param $kafka
 	 * @return array
-	 * @throws ConfigException
 	 */
-	private function kafkaConfig()
+	private function kafkaConfig($kafka)
 	{
 		$conf = new Conf();
-
-		$kafka = SConfig::get('kafka');
 		$conf->setRebalanceCb(function (KafkaConsumer $kafka, $err, array $partitions = null) {
 			if ($err == RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS) {
 				$kafka->assign($partitions);
@@ -147,18 +166,14 @@ class Kafka extends \Snowflake\Process\Process
 			}
 		});
 		$conf->set('group.id', $kafka['groupId']);
-//		$conf->set('metadata.broker.list', '127.0.0.1:2080');
-		$conf->set('metadata.broker.list', '172.26.221.220');
+		$conf->set('metadata.broker.list', $kafka['brokers']);
 		$conf->set('socket.timeout.ms', 30000);
-
 		if (function_exists('pcntl_sigprocmask')) {
 			pcntl_sigprocmask(SIG_BLOCK, array(SIGIO));
 			$conf->set('internal.termination.signal', SIGIO);
 		} else {
 			$conf->set('queue.buffering.max.ms', 1);
 		}
-
-
 		$topicConf = new TopicConf();
 		$topicConf->set('auto.commit.enable', 1);
 		$topicConf->set('auto.commit.interval.ms', 100);