From 0a9be5595a05b1c519adaea6779d496776411fda Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Mr=C2=B7x?= <as2252258@163.com>
Date: Mon, 12 Oct 2020 12:54:19 +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 | 13 +++----------
 1 file changed, 3 insertions(+), 10 deletions(-)

diff --git a/Kafka/Kafka.php b/Kafka/Kafka.php
index d49df5d0..0b3eb2c0 100644
--- a/Kafka/Kafka.php
+++ b/Kafka/Kafka.php
@@ -72,7 +72,7 @@ class Kafka extends \Snowflake\Process\Process
 				$group->add();
 				go(function () use ($group) {
 					while ([$topic, $part, $message] = $this->channel->pop()) {
-						$this->handlerExecute($group, $topic, $part, $message);
+						$this->handlerExecute($topic, $part, $message);
 					}
 				});
 			}
@@ -82,12 +82,11 @@ class Kafka extends \Snowflake\Process\Process
 
 
 	/**
-	 * @param $group
 	 * @param $topic
 	 * @param $part
 	 * @param $message
 	 */
-	protected function handlerExecute($group, $topic, $part, $message)
+	protected function handlerExecute($topic, $part, $message)
 	{
 		try {
 			$namespace = 'App\\Kafka\\' . ucfirst($topic) . 'Consumer';
@@ -98,13 +97,7 @@ class Kafka extends \Snowflake\Process\Process
 			if (!($class instanceof ConsumerInterface)) {
 				return;
 			}
-			$group->add();
-			go(function () use ($group, $class, $topic, $part, $message) {
-				defer(function () use ($group) {
-					$group->done();
-				});
-				$class->onHandler(new Struct($topic, $part, $message));
-			});
+			$class->onHandler(new Struct($topic, $part, $message));
 		} catch (\Throwable $exception) {
 			$this->application->error($exception->getMessage());
 		}