modify

System/Crontab/Consumer.php

@@ -0,0 +1,82 @@
+<?php
+
+
+namespace Snowflake\Crontab;
+
+
+use Exception;
+use Snowflake\Event;
+use Snowflake\Process\Process;
+use Snowflake\Snowflake;
+use Swoole\Coroutine;
+
+
+/**
+ * Class Consumer
+ * @package Snowflake\Crontab
+ */
+class Consumer extends Process
+{
+
+
+    /**
+     * @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());
+
+            $crontab = $redis->get('crontab:' . md5($value));
+            $redis->del('crontab:' . md5($value));
+            if (empty($crontab) || !($crontab = swoole_unserialize($crontab))) {
+                continue;
+            }
+
+            Coroutine::create(function (Crontab $value, int $startTime) {
+                $this->dispatch($value);
+            }, $crontab, $startTime);
+        }
+    }
+
+
+    /**
+     * @param Crontab $value
+     * @throws Exception
+     */
+    private function dispatch(Crontab $value)
+    {
+        $value->increment()->execute();
+        if ($value->getExecuteNumber() < $value->getMaxExecuteNumber()) {
+            $this->addTask($value);
+        } else if ($value->isLoop()) {
+            $this->addTask($value);
+        }
+        fire(Event::SYSTEM_RESOURCE_RELEASES);
+    }
+
+
+    /**
+     * @param Crontab $crontab
+     * @throws Exception
+     */
+    private 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

@@ -0,0 +1,212 @@
+<?php
+
+
+namespace Snowflake\Crontab;
+
+
+use Closure;
+use Exception;
+use Snowflake\Abstracts\BaseObject;
+use Snowflake\Event;
+use Swoole\Timer;
+
+/**
+ * Class Async
+ * @package Snowflake
+ */
+class Crontab extends BaseObject
+{
+
+
+    private array|Closure $handler;
+
+
+    private string $name = '';
+
+
+    private mixed $params;
+
+
+    private int $tickTime = 1;
+
+
+    private bool $isLoop = false;
+
+
+    private int $timerId = -1;
+
+
+    private int $max_execute_number = -1;
+
+
+    private int $execute_number = 0;
+
+
+    /**
+     * @return $this
+     */
+    public function increment(): static
+    {
+        $this->execute_number += 1;
+        return $this;
+    }
+
+    /**
+     * @return array|Closure
+     */
+    public function getHandler(): array|Closure
+    {
+        return $this->handler;
+    }
+
+    /**
+     * @return string
+     */
+    public function getName(): string
+    {
+        return $this->name;
+    }
+
+    /**
+     * @return mixed
+     */
+    public function getParams(): mixed
+    {
+        return $this->params;
+    }
+
+    /**
+     * @return int
+     */
+    public function getTickTime(): int
+    {
+        return $this->tickTime;
+    }
+
+    /**
+     * @return bool
+     */
+    public function isLoop(): bool
+    {
+        return $this->isLoop;
+    }
+
+    /**
+     * @return int
+     */
+    public function getMaxExecuteNumber(): int
+    {
+        return $this->max_execute_number;
+    }
+
+    /**
+     * @return int
+     */
+    public function getExecuteNumber(): int
+    {
+        return $this->execute_number;
+    }
+
+
+    /**
+     * @param array|Closure $handler
+     */
+    public function setHandler(array|Closure $handler): void
+    {
+        $this->handler = $handler;
+    }
+
+    /**
+     * @param string $name
+     */
+    public function setName(string $name): void
+    {
+        $this->name = $name;
+    }
+
+    /**
+     * @param mixed $params
+     */
+    public function setParams(mixed $params): void
+    {
+        $this->params = $params;
+    }
+
+    /**
+     * @param int $tickTime
+     */
+    public function setTickTime(int $tickTime): void
+    {
+        $this->tickTime = $tickTime;
+    }
+
+    /**
+     * @param bool $isLoop
+     */
+    public function setIsLoop(bool $isLoop): void
+    {
+        $this->isLoop = $isLoop;
+    }
+
+    /**
+     * @param int $max_execute_number
+     */
+    public function setMaxExecuteNumber(int $max_execute_number): void
+    {
+        $this->max_execute_number = $max_execute_number;
+    }
+
+    /**
+     * @param int $execute_number
+     */
+    public function setExecuteNumber(int $execute_number): void
+    {
+        $this->execute_number = $execute_number;
+    }
+
+    /**
+     * @return int
+     */
+    public function getTimerId(): int
+    {
+        return $this->timerId;
+    }
+
+    /**
+     * @param int $timerId
+     */
+    public function setTimerId(int $timerId): void
+    {
+        $this->timerId = $timerId;
+    }
+
+
+	/**
+	 *
+	 * @throws Exception
+	 */
+    public function clearTimer()
+    {
+        $this->warning('crontab timer clear.');
+        if (Timer::exists($this->timerId)) {
+            Timer::clear($this->timerId);
+        }
+    }
+
+
+    /**
+     * @throws Exception
+     */
+    public function execute(): void
+    {
+        try {
+            call_user_func($this->handler, $this->params, $this->name);
+        } catch (\Throwable $throwable) {
+            $this->addError($throwable->getMessage());
+        } finally {
+            fire(Event::SYSTEM_RESOURCE_RELEASES);
+        }
+    }
+
+
+}

System/Crontab/CrontabZookeeperProcess.php

@@ -0,0 +1,88 @@
+<?php
+
+
+namespace Snowflake\Crontab;
+
+
+use Exception;
+use ReflectionException;
+use Snowflake\Event;
+use Snowflake\Exception\ComponentException;
+use Snowflake\Exception\NotFindClassException;
+use Snowflake\Process\Process;
+use Snowflake\Snowflake;
+use Swoole\Coroutine;
+use Swoole\Coroutine\WaitGroup;
+use Swoole\Coroutine\Channel;
+use Swoole\Timer;
+
+/**
+ * Class CrontabZookeeperProcess
+ * @package Snowflake\Process
+ */
+class CrontabZookeeperProcess extends Process
+{
+
+
+    private Channel $channel;
+    private WaitGroup $waitGroup;
+
+
+    /** @var Crontab[] $names */
+    public array $names = [];
+
+
+    public array $scores = [];
+    public array $timers = [];
+
+
+    /**
+     * @param \Swoole\Process $process
+     * @throws ReflectionException
+     * @throws ComponentException
+     * @throws NotFindClassException
+     */
+    public function onHandler(\Swoole\Process $process): void
+    {
+        $crontab = Snowflake::app()->get('crontab');
+        $crontab->clearAll();
+
+        $process->name('Crontab zookeeper.');
+        Timer::tick(1000, function () {
+            $startTime = time();
+
+            $redis = Snowflake::app()->getRedis();
+
+            $range = $redis->zRangeByScore(Producer::CRONTAB_KEY, '0', (string)$startTime);
+            $redis->zRemRangeByScore(Producer::CRONTAB_KEY, '0', (string)$startTime);
+
+            /** @var Consumer $consumer */
+            $consumer = Snowflake::app()->get(Consumer::class);
+
+            foreach ($range as $value) {
+                $consumer->write(swoole_serialize(['crontab:' . md5($value), $startTime]));
+            }
+            $redis->release();
+        });
+    }
+
+
+    /**
+     * @param string $name
+     */
+    public function clear(string $name)
+    {
+        if (!isset($this->names[$name])) {
+            return;
+        }
+        $timers = $this->timers[$name];
+
+        $search = array_search($name, $this->scores[$timers]);
+        if ($search !== false) {
+            unset($this->scores[$timers][$search]);
+        }
+        unset($this->timers[$name], $this->names[$name]);
+    }
+
+
+}

System/Crontab/Producer.php

@@ -0,0 +1,69 @@
+<?php
+
+
+namespace Snowflake\Crontab;
+
+
+use Snowflake\Abstracts\Component;
+use Snowflake\Snowflake;
+use Exception;
+
+
+/**
+ * Class Producer
+ * @package Snowflake\Abstracts
+ */
+class Producer extends Component
+{
+
+    const CRONTAB_KEY = 'system:crontab';
+
+
+    /**
+     * @param Crontab $crontab
+     * @param $executeTime
+     * @throws Exception
+     */
+    public function dispatch(Crontab $crontab)
+    {
+        $redis = Snowflake::app()->getRedis();
+
+        $name = md5($crontab->getName());
+
+        $redis->set('crontab:' . $name, swoole_serialize($crontab));
+
+        $tickTime = time() + $crontab->getTickTime();
+
+        $redis->zAdd(self::CRONTAB_KEY, $tickTime, $crontab->getName());
+    }
+
+
+    /**
+     * @param string $name
+     * @throws Exception
+     */
+    public function clear(string $name)
+    {
+        $redis = Snowflake::app()->getRedis();
+
+        $redis->zRem(self::CRONTAB_KEY, $name);
+        $redis->del('crontab:' . md5($name));
+    }
+
+
+    /**
+     * @throws Exception
+     */
+    public function clearAll()
+    {
+        $redis = Snowflake::app()->getRedis();
+        $data = $redis->zRange(self::CRONTAB_KEY, 0, -1);
+        $redis->del(self::CRONTAB_KEY);
+        foreach ($data as $datum) {
+            $redis->del('crontab:' . md5($datum));
+        }
+        $redis->release();
+    }
+
+
+}