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