改名

Kafka/Consumer.php

@@ -60,6 +60,8 @@ class Consumer
 	 * start consumer
 	 *
 	 * @access public
+	 * @param \Closure|null $consumer
+	 * @param bool $isBlock
 	 * @return void
 	 */
 	public function start(\Closure $consumer = null, $isBlock = true)

Kafka/Kafka.php

@@ -6,10 +6,12 @@ namespace Kafka;
 
 use Snowflake\Exception\ConfigException;
 use Snowflake\Snowflake;
+use Swoole\Coroutine;
 use Swoole\Coroutine\Channel;
 use Swoole\Coroutine\WaitGroup;
 use Swoole\Process;
 use Snowflake\Abstracts\Config as SConfig;
+use function Amp\stop;
 
 /**
  * Class Queue
@@ -28,7 +30,7 @@ class Kafka extends \Snowflake\Process\Process
 	public function initConfig()
 	{
 		$kafka = SConfig::get('kafka');
-		$config = \Kafka\ConsumerConfig::getInstance();
+		$config = ConsumerConfig::getInstance();
 		$config->setMetadataRefreshIntervalMs(
 			$kafka['metadataRefreshIntervalMs'] ?? 1000
 		);
@@ -36,9 +38,6 @@ class Kafka extends \Snowflake\Process\Process
 		$config->setGroupId($kafka['groupId']);
 		$config->setBrokerVersion($kafka['version']);
 		$config->setTopics($kafka['topics']);
-
-		$this->channel = new Channel(1000);
-		go([$this, 'listener']);
 	}
 
 
@@ -49,8 +48,9 @@ class Kafka extends \Snowflake\Process\Process
 	public function onHandler(Process $process)
 	{
 		$this->initConfig();
+		$this->channelListener();
 
-		$consumer = new \Kafka\Consumer();
+		$consumer = new Consumer();
 		$consumer->setLogger(new Logger());
 		$consumer->start(function ($topic, $part, $message) {
 			$this->channel->push([$topic, $part, $message]);
@@ -61,31 +61,46 @@ class Kafka extends \Snowflake\Process\Process
 	/**
 	 * 监听通道数据传递
 	 */
-	protected function listener()
+	public function channelListener()
 	{
-		$group = new WaitGroup();
-		while ([$topic, $part, $message] = $this->channel->pop()) {
-			try {
-				$namespace = 'App\\Kafka\\' . ucfirst($topic) . 'Consumer';
-				if (!class_exists($namespace)) {
-					return;
-				}
-				$class = Snowflake::createObject($namespace);
-				if (!($class instanceof ConsumerInterface)) {
-					continue;
-				}
-				$group->add();
-				go(function () use ($group, $class, $topic, $part, $message) {
-					defer(function () use ($group) {
-						$group->done();
-					});
-					$class->onHandler(new Struct($topic, $part, $message));
-				});
-			} catch (\Throwable $exception) {
-				$this->application->error($exception->getMessage());
+		$this->channel = new Channel(1000);
+		Coroutine::create(function () {
+			$group = new WaitGroup();
+			while ([$topic, $part, $message] = $this->channel->pop()) {
+				$this->handlerExecute($group, $topic, $part, $message);
 			}
+			$group->wait();
+		});
+	}
+
+
+	/**
+	 * @param $group
+	 * @param $topic
+	 * @param $part
+	 * @param $message
+	 */
+	protected function handlerExecute($group, $topic, $part, $message)
+	{
+		try {
+			$namespace = 'App\\Kafka\\' . ucfirst($topic) . 'Consumer';
+			if (!class_exists($namespace)) {
+				return;
+			}
+			$class = Snowflake::createObject($namespace);
+			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));
+			});
+		} catch (\Throwable $exception) {
+			$this->application->error($exception->getMessage());
 		}
-		$group->wait();
 	}