modify

HttpServer/Events/OnPipeMessage.php

@@ -4,15 +4,11 @@ declare(strict_types=1);
 namespace HttpServer\Events;
 
 
-use Annotation\Loader;
 use Exception;
 use HttpServer\Abstracts\Callback;
-use Kafka\ConsumerInterface;
 use Kafka\Struct;
-use Kafka\TaskContainer;
 use Snowflake\Event;
 use Snowflake\Exception\ComponentException;
-use Snowflake\Snowflake;
 use Swoole\Server;
 
 /**
@@ -35,6 +31,7 @@ class OnPipeMessage extends Callback
             $swoole_unserialize = swoole_unserialize($message);
             match ($swoole_unserialize['action'] ?? null) {
                 'kafka' => $this->onKafkaWorker($swoole_unserialize),
+                'crontab' => $this->onCrontabWorker($swoole_unserialize),
                 default => $this->onMessageWorker($server, $src_worker_id, $message)
             };
         } catch (\Throwable $exception) {
@@ -45,6 +42,19 @@ class OnPipeMessage extends Callback
     }
 
 
+    /**
+     * @param array $message
+     * @return string
+     */
+    private function onCrontabWorker(array $message)
+    {
+        /** @var \Snowflake\Crontab\Crontab $crontab */
+        $crontab = $message['handler'] ?? null;
+        $crontab->increment()->execute();
+        return 'success';
+    }
+
+
     /**
      * @param $server
      * @param $src_worker_id

Kafka/TaskContainer.php

@@ -48,7 +48,6 @@ class TaskContainer extends BaseObject
     public function process($topic, Struct $struct)
     {
         $handler = $this->_topics[$topic] ?? null;
-        var_dump($this->_topics, $topic, $struct);
         if (empty($handler)) {
             return;
         }

System/Crontab/Consumer.php

@@ -1,129 +0,0 @@
-<?php
-
-
-namespace Snowflake\Crontab;
-
-
-use Exception;
-use ReflectionException;
-use Snowflake\Event;
-use Snowflake\Exception\ComponentException;
-use Snowflake\Exception\ConfigException;
-use Snowflake\Exception\NotFindClassException;
-use Snowflake\Process\Process;
-use Snowflake\Snowflake;
-use Swoole\Coroutine;
-
-
-/**
- * Class Consumer
- * @package Snowflake\Crontab
- */
-class Consumer extends Process
-{
-
-    public Coroutine\Channel $channel;
-
-
-    /**
-     * @param \Swoole\Process $process
-     * @throws Exception
-     */
-    public function onHandler(\Swoole\Process $process): void
-    {
-        if (Snowflake::getPlatform()->isLinux()) {
-            name($this->pid, 'Crontab consumer');
-        }
-
-        $this->channel = new Coroutine\Channel(2000);
-        go(function () {
-            $this->popChannel();
-        });
-
-        $this->recovery();
-        $this->tick($process);
-    }
-
-
-    private function recovery()
-    {
-        $redis = redis()->hGetAll('crontab:wait:execute');
-        foreach ($redis as $redi) {
-            $this->channel->push($redi);
-        }
-        fire(Event::SYSTEM_RESOURCE_RELEASES);
-    }
-
-
-    /**
-     * @throws Exception
-     */
-    public function popChannel()
-    {
-        /** @var Crontab $crontab */
-        $crontab = $this->channel->pop(-1);
-        go(function () use ($crontab) {
-            try {
-                $crontab->increment()->execute();
-                if ($crontab->getExecuteNumber() < $crontab->getMaxExecuteNumber()) {
-                    Consumer::addTask($crontab);
-                } else if ($crontab->isLoop()) {
-                    Consumer::addTask($crontab);
-                }
-            } catch (\Throwable $throwable) {
-                logger()->addError($throwable, 'throwable');
-            } finally {
-                fire(Event::SYSTEM_RESOURCE_RELEASES);
-            }
-        });
-        $this->popChannel();
-    }
-
-
-    /**
-     * @param \Swoole\Process $process
-     * @throws ReflectionException
-     * @throws ComponentException
-     * @throws ConfigException
-     * @throws NotFindClassException
-     * @throws Exception
-     */
-    public function tick(\Swoole\Process $process)
-    {
-        $value = $process->read(40);
-
-        $redis = Snowflake::app()->getRedis();
-
-        /** @var Crontab $crontab */
-        $crontab = swoole_unserialize($redis->get($value));
-        $redis->hSet('crontab:wait:execute', $crontab->getName(), $redis->get($value));
-        $redis->del($value);
-        if (is_object($crontab)) {
-            $this->channel->push($crontab);
-        }
-
-        $redis->release();
-
-        $this->tick($process);
-    }
-
-
-    /**
-     * @param Crontab $crontab
-     * @throws Exception
-     */
-    private static function addTask(Crontab $crontab)
-    {
-        $redis = Snowflake::app()->getRedis();
-
-        $name = md5($crontab->getName());
-
-        $redis->set('crontab:' . $name, swoole_serialize($crontab));
-
-        $tickTime = time() + $crontab->getTickTime();
-
-        $redis->zAdd(Producer::CRONTAB_KEY, $tickTime, $crontab->getName());
-    }
-
-
-}

System/Crontab/Crontab.php

@@ -8,6 +8,7 @@ use Closure;
 use Exception;
 use Snowflake\Abstracts\BaseObject;
 use Snowflake\Event;
+use Snowflake\Snowflake;
 use Swoole\Timer;
 
 /**
@@ -17,6 +18,7 @@ use Swoole\Timer;
 class Crontab extends BaseObject
 {
 
+    const WAIT_END = 'crontab:wait:execute';
 
     private array|Closure $handler;
 
@@ -194,16 +196,56 @@ class Crontab extends BaseObject
     }
 
 
+    /**
+     * @throws \Exception
+     */
+    private function recover()
+    {
+        $redis = Snowflake::app()->getRedis();
+
+        $redis->set('crontab:' . md5($this->getName()), swoole_serialize($this));
+
+        $tickTime = time() + $this->getTickTime();
+
+        $redis->zAdd(Producer::CRONTAB_KEY, $tickTime, $this->getName());
+    }
+
+
     /**
      * @throws Exception
      */
     public function execute(): void
     {
+        try {
+            $redis = Snowflake::app()->getRedis();
+
+            $name_md5 = md5($this->getName());
+
+            $redis->hSet(self::WAIT_END, $name_md5, serialize($this));
             $params = call_user_func($this->handler, $this->params, $this->name);
-        \redis()->hDel('crontab:wait:execute', $this->getName());
+            $redis->hDel(self::WAIT_END, $name_md5);
-        if ($params !== null) {
+            if ($params === null) {
+                return;
+            }
             $name = date('Y_m_d_H_i_s.' . $this->name . '.log');
             write(storage($name, '/log/crontab'), serialize($params));
+        } catch (\Throwable $throwable) {
+            logger()->addError($throwable, 'throwable');
+        } finally {
+            $this->after();
+        }
+    }
+
+
+    /**
+     * @throws \Exception
+     */
+    private function after(): void
+    {
+        if ($this->getExecuteNumber() < $this->getMaxExecuteNumber()) {
+            $this->recover();
+        } else if ($this->isLoop()) {
+            $this->recover();
         }
     }
 

System/Crontab/CrontabProviders.php

@@ -37,8 +37,7 @@ class CrontabProviders extends Providers
         if (Config::get('crontab.enable') !== true) {
             return;
         }
-        $server->addProcess('CrontabZookeeper', ZookeeperProcess::class);
+        $server->addProcess('CrontabZookeeper', Zookeeper::class);
-        $server->addProcess('Consumer', Consumer::class);
     }
 
 }

System/Crontab/Zookeeper.php

@@ -12,10 +12,10 @@ use Swoole\Coroutine\Channel;
 use Swoole\Timer;
 
 /**
- * Class ZookeeperProcess
+ * Class Zookeeper
  * @package Snowflake\Process
  */
-class ZookeeperProcess extends Process
+class Zookeeper extends Process
 {
 
 
@@ -39,31 +39,52 @@ class ZookeeperProcess extends Process
     {
         $crontab = Snowflake::app()->get('crontab');
         $crontab->clearAll();
-
         if (Snowflake::getPlatform()->isLinux()) {
             name($this->pid, 'Crontab zookeeper.');
         }
         Timer::tick(1000, function () {
-            $startTime = time();
+            [$range, $redis] = $this->loadCrotabTask();
+            $server = Snowflake::app()->getSwoole();
+            $setting = $server->setting['worker_num'];
+            foreach ($range as $value) {
+                $this->dispatch($server, $redis, $setting, $value);
+            }
+            $redis->release();
+        });
+    }
 
+
+    /**
+     * @param $server
+     * @param $redis
+     * @throws \Exception
+     */
+    private function dispatch($server, $redis, $setting, $value)
+    {
+        $server->sendMessage(swoole_serialize([
+            'action' => 'crontab', 'handler' => swoole_unserialize($redis->get('crontab:' . md5($value)))
+        ]), random_int(0, $setting - 1));
+        $redis->del('crontab:' . md5($value));
+    }
+
+
+    /**
+     * @return array
+     * @throws \Exception
+     */
+    private function loadCrotabTask()
+    {
         $redis = Snowflake::app()->getRedis();
 
+        $startTime = time();
+
         $redis->multi();
         $range = $redis->zRangeByScore(Producer::CRONTAB_KEY, '0', (string)$startTime);
         $redis->zRemRangeByScore(Producer::CRONTAB_KEY, '0', (string)$startTime);
         $redis->exec();
 
-            $redis->release();
+        return [$range, $redis];
-
-            /** @var Consumer $consumer */
-            $consumer = Snowflake::app()->get(Consumer::class);
-
-            foreach ($range as $value) {
-                $consumer->write('crontab:' . md5($value));
     }
-        });
-    }
-
 
 
     /**