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 @@ + 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(); }