From 4d6f7d1d13f90caf6d2cae7f9f5bf5cbf5692581 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E5=90=91=E6=9E=97?= <as2252258@163.com>
Date: Thu, 3 Feb 2022 14:54:38 +0800
Subject: [PATCH] =?UTF-8?q?Revert=20"=E6=94=B9=E5=90=8D"?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

This reverts commit fdf58326
---
 kiri-engine/Abstracts/BaseApplication.php |  6 +--
 kiri-server/CoroutineTasker.php           | 54 +++++++++++++++++++++++
 kiri-task/CoroutineTaskExecute.php        |  5 ++-
 3 files changed, 60 insertions(+), 5 deletions(-)
 create mode 100644 kiri-server/CoroutineTasker.php

diff --git a/kiri-engine/Abstracts/BaseApplication.php b/kiri-engine/Abstracts/BaseApplication.php
index 94d9bb37..3658c667 100644
--- a/kiri-engine/Abstracts/BaseApplication.php
+++ b/kiri-engine/Abstracts/BaseApplication.php
@@ -12,19 +12,19 @@ namespace Kiri\Abstracts;
 
 use Database\Connection;
 use Exception;
-use Kiri\Message\Handler\Router;
 use Kafka\KafkaProvider;
-use Kiri\Async;
 use Kiri;
 use Kiri\Annotation\Annotation as SAnnotation;
+use Kiri\Async;
 use Kiri\Cache\Redis;
 use Kiri\Di\LocalService;
 use Kiri\Error\{ErrorHandler, Logger};
 use Kiri\Exception\{InitException, NotFindClassException};
-use ReflectionException;
+use Kiri\Message\Handler\Router;
 use Kiri\Server\{Server, ServerManager};
 use Kiri\Task\AsyncTaskExecute;
 use Kiri\Task\OnTaskInterface;
+use ReflectionException;
 use Swoole\Table;
 
 /**
diff --git a/kiri-server/CoroutineTasker.php b/kiri-server/CoroutineTasker.php
new file mode 100644
index 00000000..500101cc
--- /dev/null
+++ b/kiri-server/CoroutineTasker.php
@@ -0,0 +1,54 @@
+<?php
+
+namespace Kiri\Server;
+
+use Kiri\Abstracts\Component;
+use Kiri\Abstracts\Config;
+use Kiri\Exception\ConfigException;
+use Kiri\Task\OnServerTask;
+use Swoole\Coroutine;
+use Swoole\Coroutine\Channel;
+use Swoole\Server\Task;
+
+class CoroutineTasker extends Component
+{
+
+
+	public Channel $channel;
+
+
+	/**
+	 * @return void
+	 * @throws ConfigException
+	 */
+	public function initCore()
+	{
+		$taskWorker = Config::get('server.settings.task_worker_num', 0);
+
+		if ($taskWorker > 1) {
+			$this->channel = new Channel($taskWorker);
+			for ($i = 0; $i < $taskWorker; $i++) {
+
+				Coroutine::create(function () {
+					while ($this->channel->capacity) {
+						$data = $this->channel->pop(-1);
+
+						$execute = $this->getContainer()->get(OnServerTask::class);
+
+						$task = new Task();
+						$task->data = $data;
+						$execute->onCoroutineTask(null, $task);
+					}
+				});
+			}
+		}
+	}
+
+
+	public function dispatch($data)
+	{
+		$this->channel->push($data);
+	}
+
+
+}
diff --git a/kiri-task/CoroutineTaskExecute.php b/kiri-task/CoroutineTaskExecute.php
index 42f8d554..cf6f66e0 100644
--- a/kiri-task/CoroutineTaskExecute.php
+++ b/kiri-task/CoroutineTaskExecute.php
@@ -56,14 +56,15 @@ class CoroutineTaskExecute extends Component
 	 */
 	protected function handler()
 	{
+		Coroutine\defer(function () {
+			$this->handler();
+		});
 		$data = $this->channel->pop(-1);
 
 		$task = new Task();
 		$task->data = $data;
 
 		$this->taskServer->onCoroutineTask(null, $task);
-
-		$this->handler();
 	}