From 8217f670fe5781460340f09a3c2fb0e59740e64c Mon Sep 17 00:00:00 2001
From: "as2252258@163.com" <as22522582>
Date: Fri, 26 Mar 2021 01:39:22 +0800
Subject: [PATCH] modify

---
 System/Crontab/Consumer.php | 57 ++++++++++++++++++++++++++++++-------
 1 file changed, 46 insertions(+), 11 deletions(-)

diff --git a/System/Crontab/Consumer.php b/System/Crontab/Consumer.php
index 1cae3d45..616315c6 100644
--- a/System/Crontab/Consumer.php
+++ b/System/Crontab/Consumer.php
@@ -8,6 +8,7 @@ use Exception;
 use Snowflake\Event;
 use Snowflake\Process\Process;
 use Snowflake\Snowflake;
+use Swoole\Coroutine;
 
 
 /**
@@ -17,27 +18,61 @@ use Snowflake\Snowflake;
 class Consumer extends Process
 {
 
+    public Coroutine\Channel $channel;
+
 
     /**
      * @param \Swoole\Process $process
      */
     public function onHandler(\Swoole\Process $process): void
     {
-        // TODO: Implement onHandler() method.
-        $redis = Snowflake::app()->getRedis();
-
         $process->name('Crontab consumer');
 
-        while (true) {
-            [$value, $startTime] = swoole_unserialize($process->read());
+        $this->channel = new Coroutine\Channel(2000);
+        Coroutine\go(function () {
+            $this->popChannel();
+        });
+        $this->tick($process);
+    }
 
-            $crontab = swoole_unserialize($redis->get($value));
-            $redis->del($value);
-            if (!is_object($crontab)) {
-                continue;
-            }
-            $this->dispatch($crontab);
+
+    /**
+     * @throws Exception
+     */
+    public function popChannel()
+    {
+        $crontab = $this->channel->pop(-1);
+
+        $this->dispatch($crontab);
+
+        $this->popChannel();
+    }
+
+
+    /**
+     * @param \Swoole\Process $process
+     * @throws \ReflectionException
+     * @throws \Snowflake\Exception\ComponentException
+     * @throws \Snowflake\Exception\ConfigException
+     * @throws \Snowflake\Exception\NotFindClassException
+     */
+    public function tick(\Swoole\Process $process)
+    {
+        [$value, $startTime] = swoole_unserialize($process->read(-1));
+
+        $redis = Snowflake::app()->getRedis();
+
+        $crontab = swoole_unserialize($redis->get($value));
+        $redis->del($value);
+        if (is_object($crontab)) {
+            $this->channel->push($crontab);
         }
+
+        $redis->release();
+
+        Coroutine::sleep(0.05);
+
+        $this->tick($process);
     }