改名

Kafka/Kafka.php

@@ -6,6 +6,8 @@ namespace Kafka;
 
 use Snowflake\Exception\ConfigException;
 use Snowflake\Snowflake;
+use Swoole\Coroutine\Channel;
+use Swoole\Coroutine\WaitGroup;
 use Swoole\Process;
 use Snowflake\Abstracts\Config as SConfig;
 
@@ -16,11 +18,14 @@ use Snowflake\Abstracts\Config as SConfig;
 class Kafka extends \Snowflake\Process\Process
 {
 
+	/** @var Channel */
+	protected $channel;
+
+
 	/**
-	 * @param Process $process
 	 * @throws ConfigException
 	 */
-	public function onHandler(Process $process)
+	public function initConfig()
 	{
 		$kafka = SConfig::get('kafka');
 		$config = \Kafka\ConsumerConfig::getInstance();
@@ -32,23 +37,56 @@ class Kafka extends \Snowflake\Process\Process
 		$config->setBrokerVersion($kafka['version']);
 		$config->setTopics($kafka['topics']);
 
+		$this->channel = new Channel(1000);
+		$this->initCoroutine();
+	}
+
+
+	/**
+	 * @param Process $process
+	 * @throws ConfigException
+	 */
+	public function onHandler(Process $process)
+	{
+		$this->initConfig();
+
 		$consumer = new \Kafka\Consumer();
 		$consumer->setLogger(new Logger());
 		$consumer->start(function ($topic, $part, $message) {
+			$this->channel->push([$topic, $part, $message]);
+		});
+	}
+
+
+	/**
+	 *
+	 */
+	public function initCoroutine()
+	{
+		$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) {
-					$class->onHandler(new Struct($topic, $part, $message));
+				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());
 			}
-		});
-
+		}
+		$group->wait();
 	}
 
+
 }