From 3cd82aad2d822371066f1d1089d17ef82a589325 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mr=C2=B7x?= Date: Sat, 20 Feb 2021 15:35:12 +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 | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/Kafka/Kafka.php b/Kafka/Kafka.php index b844f59f..2fcc8bf7 100644 --- a/Kafka/Kafka.php +++ b/Kafka/Kafka.php @@ -114,9 +114,7 @@ class Kafka extends \Snowflake\Process\Process defer(function () use ($group) { $group->done(); }); - while ($messages = $this->channel->pop()) { - $this->handlerExecute($messages[0], $messages[1]); - } + $this->popMessage(); }); } $group->wait(); @@ -124,6 +122,19 @@ class Kafka extends \Snowflake\Process\Process } + /** + * 优化执行方式 + */ + public function popMessage() + { + $messages = $this->channel->pop(-1); + if (!empty($messages)) { + $this->handlerExecute($messages[0], $messages[1]); + } + $this->popMessage(); + } + + /** * @param $topic * @param $message