modify

2021-03-20 02:33:50 +08:00
parent 25a09a379a
commit 06422d5073
3 changed files with 253 additions and 231 deletions
@@ -4,6 +4,8 @@
 namespace Snowflake\Abstracts;


+use Snowflake\Core\Json;
+use Snowflake\Process\CrontabProcess;
 use Snowflake\Snowflake;
 use Exception;

@@ -16,51 +18,40 @@ class Crontab extends Component
 {


-	/**
-	 * @param \Snowflake\Crontab $crontab
-	 * @param $executeTime
-	 * @throws Exception
-	 */
-	public function dispatch(\Snowflake\Crontab $crontab, int $executeTime)
-	{
-		$redis = Snowflake::app()->getRedis();
-
-		$redis->zAdd('system:crontab', (string)$executeTime, serialize($crontab));
-	}
+    /**
+     * @param \Snowflake\Crontab $crontab
+     * @param $executeTime
+     * @throws Exception
+     */
+    public function dispatch(\Snowflake\Crontab $crontab)
+    {
+        /** @var CrontabProcess $redis */
+        $redis = Snowflake::app()->get(CrontabProcess::class);
+        $redis->write(serialize($crontab));
+    }


-	/**
-	 * @param string $name
-	 * @throws Exception
-	 */
-	public function clear(string $name)
-	{
-		$redis = Snowflake::app()->getRedis();
-
-		$data = $redis->zRevRange('system:crontab', 0, -1);
-		if (empty($data)) {
-			return;
-		}
-		foreach ($data as $datum) {
-			/** @var \Snowflake\Crontab $crontab */
-			$crontab = unserialize($datum);
-			if ($crontab->getName() == $name) {
-				$redis->zRem('system:crontab', $datum);
-			}
-		}
-
-	}
+    /**
+     * @param string $name
+     * @throws Exception
+     */
+    public function clear(string $name)
+    {
+        /** @var CrontabProcess $redis */
+        $redis = Snowflake::app()->get(CrontabProcess::class);
+        $redis->write(Json::encode(['action' => 'clear', 'name' => $name]));
+    }


-	/**
-	 * @throws Exception
-	 */
-	public function clearAll()
-	{
-		$redis = Snowflake::app()->getRedis();
-
-		$redis->del('system:crontab');
-	}
+    /**
+     * @throws Exception
+     */
+    public function clearAll()
+    {
+        /** @var CrontabProcess $redis */
+        $redis = Snowflake::app()->get(CrontabProcess::class);
+        $redis->write(Json::encode(['action' => 'clearAll']));
+    }


 }
@@ -7,6 +7,7 @@ namespace Snowflake;
 use Closure;
 use Exception;
 use Snowflake\Abstracts\BaseObject;
+use Swoole\Timer;

 /**
 * Class Async
@@ -16,163 +17,189 @@ class Crontab extends BaseObject
 {


-	private array|Closure $handler;
+    private array|Closure $handler;


-	private string $name = '';
+    private string $name = '';


-	private mixed $params;
+    private mixed $params;


-	private int $tickTime = 1;
+    private int $tickTime = 1;


-	private bool $isLoop = false;
+    private bool $isLoop = false;


-	private int $max_execute_number = -1;
+    private int $timerId = -1;


-	private int $execute_number = 0;
-
-	/**
-	 * @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;
-	}
+    private int $max_execute_number = -1;


-	/**
-	 * @param array|Closure $handler
-	 */
-	public function setHandler(array|Closure $handler): void
-	{
-		$this->handler = $handler;
-	}
+    private int $execute_number = 0;

-	/**
-	 * @param string $name
-	 */
-	public function setName(string $name): void
-	{
-		$this->name = $name;
-	}
+    /**
+     * @return array|Closure
+     */
+    public function getHandler(): array|Closure
+    {
+        return $this->handler;
+    }

-	/**
-	 * @param mixed $params
-	 */
-	public function setParams(mixed $params): void
-	{
-		$this->params = $params;
-	}
+    /**
+     * @return string
+     */
+    public function getName(): string
+    {
+        return $this->name;
+    }

-	/**
-	 * @param int $tickTime
-	 */
-	public function setTickTime(int $tickTime): void
-	{
-		$this->tickTime = $tickTime;
-	}
+    /**
+     * @return mixed
+     */
+    public function getParams(): mixed
+    {
+        return $this->params;
+    }

-	/**
-	 * @param bool $isLoop
-	 */
-	public function setIsLoop(bool $isLoop): void
-	{
-		$this->isLoop = $isLoop;
-	}
+    /**
+     * @return int
+     */
+    public function getTickTime(): int
+    {
+        return $this->tickTime;
+    }

-	/**
-	 * @param int $max_execute_number
-	 */
-	public function setMaxExecuteNumber(int $max_execute_number): void
-	{
-		$this->max_execute_number = $max_execute_number;
-	}
+    /**
+     * @return bool
+     */
+    public function isLoop(): bool
+    {
+        return $this->isLoop;
+    }

-	/**
-	 * @param int $execute_number
-	 */
-	public function setExecuteNumber(int $execute_number): void
-	{
-		$this->execute_number = $execute_number;
-	}
+    /**
+     * @return int
+     */
+    public function getMaxExecuteNumber(): int
+    {
+        return $this->max_execute_number;
+    }
+
+    /**
+     * @return int
+     */
+    public function getExecuteNumber(): int
+    {
+        return $this->execute_number;
+    }


-	/**
-	 * @throws Exception
-	 */
-	public function execute(): void
-	{
-		$redis = Snowflake::app()->getRedis();
-		try {
-			$this->execute_number += 1;
-			call_user_func($this->handler, $this->params);
-			if ($this->isLoop === false) {
-				return;
-			}
-			if ($this->max_execute_number === -1) {
-				$redis->zAdd('system:crontab', time() + $this->tickTime, serialize($this));
-			} else if ($this->execute_number < $this->max_execute_number) {
-				$redis->zAdd('system:crontab', time() + $this->tickTime, serialize($this));
-			}
-		} catch (\Throwable $throwable) {
-			$this->addError($throwable->getMessage());
-		} finally {
-			$redis->release();
-		}
-	}
+    /**
+     * @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;
+    }
+
+
+    /**
+     *
+     */
+    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->execute_number += 1;
+
+            if ($this->execute_number < $this->max_execute_number) {
+                $this->clearTimer();
+            }
+        } catch (\Throwable $throwable) {
+            $this->addError($throwable->getMessage());
+        } finally {
+            fire(Event::SYSTEM_RESOURCE_RELEASES);
+        }
+    }


 }
@@ -5,6 +5,7 @@ namespace Snowflake\Process;


 use ReflectionException;
+use Snowflake\Core\Json;
 use Snowflake\Crontab;
 use Snowflake\Exception\ComponentException;
 use Snowflake\Exception\ConfigException;
@@ -23,67 +24,70 @@ use Swoole\Timer;
 class CrontabProcess extends Process
 {

-
-	public Channel $channel;
+    /** @var Crontab[] $names */
+    public array $names = [];


-	/**
-	 * @param \Swoole\Process $process
-	 */
-	public function onHandler(\Swoole\Process $process): void
-	{
-		$this->channel = new Channel(5000);
-
-		Coroutine::create([$this, 'execute']);
-
-		Timer::tick(1000, [$this, 'systemLoop']);
-	}
+    /**
+     * @param \Swoole\Process $process
+     */
+    public function onHandler(\Swoole\Process $process): void
+    {
+        while (true) {
+            $content = $process->read();
+            $_content = json_decode($content, true);
+            if (is_null($_content)) {
+                $this->jobDelivery($content);
+            } else {
+                $this->otherAction($content);
+            }
+        }
+    }


-	/**
-	 * @throws \Exception
-	 */
-	public function execute()
-	{
-		while (true) {
-			/** @var Crontab $list */
-			$list = $this->channel->pop(-1);
-			$list->execute();
-		}
-	}
+    /**
+     * @param $content
+     */
+    private function otherAction($content)
+    {
+        call_user_func(match ($content['action']) {
+            'clear' => function ($content) {
+                if (!isset($this->names[$content['name']])) {
+                    return;
+                }
+                $this->names[$content['name']]->clearTimer();
+            },
+            'clearAll' => function () {
+                foreach ($this->names as $name => $crontab) {
+                    $crontab->clearTimer();
+                }
+            },
+            default => function () {
+                $this->application->error('unknown action');
+            }
+        });
+    }


-	/**
-	 * @throws ReflectionException
-	 * @throws ComponentException
-	 * @throws ConfigException
-	 * @throws NotFindClassException
-	 * @throws Exception
-	 * @throws \Exception
-	 */
-	public function systemLoop()
-	{
-		$score = time();
-		$redis = Snowflake::app()->getRedis();
+    /**
+     * @param $content
+     */
+    private function jobDelivery($content)
+    {
+        $content = unserialize($content);
+        $this->names[$content->getName()] = $content;
+        if (!($content instanceof Crontab)) {
+            return;
+        }
+        $runTicker = [$content, 'execute'];
+        $timer = $content->getTickTime() * 1000;
+        if ($content->isLoop()) {
+            $timerId = Timer::tick($timer, $runTicker);
+        } else {
+            $timerId = Timer::after($timer, $runTicker);
+        }
+        $content->setTimerId($timerId);
+    }

-		$lists = $redis->zRangeByScore('system:crontab', '0', (string)$score);
-		$redis->zRemRangeByScore('system:crontab', '0', (string)$score);
-
-		if (empty($lists)) {
-			$redis->release();
-			return;
-		}
-
-		$barrier = Barrier::make();
-		foreach ($lists as $list) {
-			$list = unserialize($list);
-			if (!($list instanceof Crontab)) {
-				continue;
-			}
-			$this->channel->push($list);
-		}
-		Barrier::wait($barrier);
-		$redis->release();
-	}

 }