diff --git a/HttpServer/Server.php b/HttpServer/Server.php index 403aba36..7e0ad18c 100644 --- a/HttpServer/Server.php +++ b/HttpServer/Server.php @@ -62,6 +62,20 @@ class Server extends Application public $daemon = 0; + + private $process = []; + + + /** + * @param $name + * @param $process + */ + public function addProcess($name, $process) + { + $this->process[$name] = $process; + } + + /** * @param array $configs * @return Http|Packet|Receive|Websocket @@ -172,7 +186,6 @@ class Server extends Application } - /** * @throws ConfigException * @throws Exception @@ -180,11 +193,23 @@ class Server extends Application public function onProcessListener() { $processes = Config::get('processes'); + if (!empty($processes) && is_array($processes)) { + return $this->deliveryProcess(merge($processes, $this->process)); + } + return $this->deliveryProcess($this->process); + } + + + /** + * @param $processes + * @throws Exception + */ + private function deliveryProcess($processes) + { + $application = Snowflake::app(); if (empty($processes) || !is_array($processes)) { return; } - - $application = Snowflake::app(); foreach ($processes as $name => $process) { $this->debug(sprintf('Process %s', $process)); if (!is_string($process)) { diff --git a/Queue/Abstracts/AbstractsQueue.php b/Queue/Abstracts/AbstractsQueue.php new file mode 100644 index 00000000..b25e05ec --- /dev/null +++ b/Queue/Abstracts/AbstractsQueue.php @@ -0,0 +1,16 @@ +getRedis(); + $serialize = serialize($consumer); + if (!$redis->lock($hash = md5($serialize))) { + return false; + } + $isExists = $redis->zRevRank($key, $serialize); + if ($isExists !== null) { + $redis->zAdd($key, $score, $serialize); + } + return $redis->unlink($hash); + } + + + /** + * @param $key + * @param Consumer $consumer + * @return false|int + * @throws ComponentException + * @throws Exception + */ + protected function pop($key, Consumer $consumer) + { + $serialize = serialize($consumer); + $redis = Snowflake::app()->getRedis(); + if (!$redis->lock($hash = md5($serialize))) { + return false; + } + $isExists = $redis->zRevRank($key, $serialize); + if ($isExists === null) { + return $redis->unlink($hash); + } + $redis->zRem($key, $serialize); + return $redis->unlink($hash); + } + + +} diff --git a/Queue/Abstracts/Relyon.php b/Queue/Abstracts/Relyon.php new file mode 100644 index 00000000..6a8131ce --- /dev/null +++ b/Queue/Abstracts/Relyon.php @@ -0,0 +1,32 @@ +push(self::QUEUE_COMPLETE, $consumer, $score); + } + + + /** + * @param $consumer + * @return false|int + * @throws ComponentException + */ + public function del(Consumer $consumer) + { + return $this->pop(self::QUEUE_COMPLETE, $consumer); + } + + +} diff --git a/Queue/Consumer.php b/Queue/Consumer.php new file mode 100644 index 00000000..7dd2c02b --- /dev/null +++ b/Queue/Consumer.php @@ -0,0 +1,24 @@ +waiting = Snowflake::createObject(Waiting::class); + $this->running = Snowflake::createObject(Running::class); + $this->complete = Snowflake::createObject(Complete::class); + + Process::signal(9 | 15, function () { + $this->shutdown = true; + }); + } + + + /** + * @param Process $process + * @throws ComponentException + */ + public function onHandler(Process $process) + { + $redis = Snowflake::app()->getRedis(); + Timer::tick(50, function ($timerId) use ($redis) { + if ($this->shutdown) { + return Timer::clear($timerId); + } + $data = $redis->zRevRangeByScore(Waiting::QUEUE_WAITING, 0, 20); + if (empty($data)) { + return 1; + } + return $this->scheduler($data); + }); + } + + + /** + * @param Consumer $consumer + * @param int $score + * @throws ComponentException + */ + public function delivery(Consumer $consumer, $score = 0) + { + $this->waiting->add($consumer, $score); + } + + + /** + * @param array $data + * @throws Exception + */ + private function scheduler($data) + { + $scheduler = new Coroutine\Scheduler(); + foreach ($data as $datum) { + $scheduler->add([$this, 'runner'], $datum); + } + $scheduler->start(); + if ($this->shutdown === true) { + Snowflake::shutdown($this); + } + } + + + /** + * @param $class + * @return mixed|void + * @throws ComponentException + */ + private function runner(string $class) + { + $logger = $this->application->getLogger(); + try { + $rely_on = unserialize($class); + $this->waiting->remove($class); + if ($rely_on instanceof Consumer) { + return; + } + $this->running->add($rely_on); + $rely_on->onRunning(); + } catch (\Throwable $exception) { + $logger->write($exception->getMessage(), 'queue'); + } finally { + $this->running->del($class); + if (isset($rely_on) && $rely_on instanceof Consumer) { + $rely_on->onComplete(); + $this->complete->add($rely_on); + } + } + } + + + /** + * @return Waiting + */ + public function getWaiting(): Waiting + { + return $this->waiting; + } + + /** + * @return Complete + */ + public function getComplete(): Complete + { + return $this->complete; + } + + /** + * @return Running + */ + public function getRunning(): Running + { + return $this->running; + } + +} diff --git a/Queue/QueueProviders.php b/Queue/QueueProviders.php new file mode 100644 index 00000000..f00c9c81 --- /dev/null +++ b/Queue/QueueProviders.php @@ -0,0 +1,31 @@ +get('server'); + $server->addProcess('queue', Queue::class); + } + +} diff --git a/Queue/Running.php b/Queue/Running.php new file mode 100644 index 00000000..5fa1472a --- /dev/null +++ b/Queue/Running.php @@ -0,0 +1,51 @@ +push(self::QUEUE_RUNNING, $consumer, $score); + } + + + /** + * @param $consumer + * @return false|int + * @throws ComponentException + */ + public function del(string $consumer) + { + $consumer = unserialize($consumer); + if (!($consumer instanceof Consumer)) { + return true; + } + return $this->pop(self::QUEUE_RUNNING, $consumer); + } + + +} diff --git a/Queue/TestQueue.php b/Queue/TestQueue.php new file mode 100644 index 00000000..04137ee4 --- /dev/null +++ b/Queue/TestQueue.php @@ -0,0 +1,48 @@ +params = $params; + $this->onWaiting(); + } + + /** + * + */ + public function onWaiting() + { + + + + } + + public function onRunning() + { + // TODO: Implement onRunning() method. + } + + public function onComplete() + { + // TODO: Implement onComplete() method. + } +} diff --git a/Queue/Waiting.php b/Queue/Waiting.php new file mode 100644 index 00000000..cf5e9df9 --- /dev/null +++ b/Queue/Waiting.php @@ -0,0 +1,45 @@ +push(self::QUEUE_WAITING, $consumer, $score); + } + + + /** + * @param $consumer + * @return false|int + * @throws ComponentException + */ + public function del(Consumer $consumer) + { + return $this->pop(self::QUEUE_WAITING, $consumer); + } + +} diff --git a/System/Application.php b/System/Application.php index 3014caae..3d96e15f 100644 --- a/System/Application.php +++ b/System/Application.php @@ -14,10 +14,12 @@ use Console\ConsoleProviders; use Database\DatabasesProviders; use Exception; use HttpServer\ServerProviders; +use Queue\QueueProviders; use Snowflake\Abstracts\BaseApplication; use Snowflake\Abstracts\Config; use Snowflake\Abstracts\Input; use Snowflake\Exception\NotFindClassException; +use Snowflake\Exception\ConfigException; use Swoole\Timer; /** @@ -37,6 +39,7 @@ class Application extends BaseApplication /** + * @throws ConfigException * @throws NotFindClassException */ public function init() @@ -44,6 +47,9 @@ class Application extends BaseApplication $this->import(ConsoleProviders::class); $this->import(DatabasesProviders::class); $this->import(ServerProviders::class); + if (Config::get('queue.enable', false, false)) { + $this->import(QueueProviders::class); + } } diff --git a/System/Error/Logger.php b/System/Error/Logger.php index da6a95fc..5ed13831 100644 --- a/System/Error/Logger.php +++ b/System/Error/Logger.php @@ -142,7 +142,7 @@ class Logger extends Component /** * @param $messages * @param string $category - * @throws Exception + * @throws */ public function write(string $messages, $category = 'app') { diff --git a/System/Snowflake.php b/System/Snowflake.php index 655544c8..d57ec265 100644 --- a/System/Snowflake.php +++ b/System/Snowflake.php @@ -10,6 +10,7 @@ use ReflectionException; use Snowflake\Abstracts\Config; use Snowflake\Di\Container; use Snowflake\Exception\NotFindClassException; +use Snowflake\Process\Process; use Swoole\Coroutine; class Snowflake @@ -214,6 +215,23 @@ class Snowflake } + /** + * @param $process + * @return mixed|void + */ + public static function shutdown($process) + { + static::app()->server->getServer()->shutdown(); + if ($process instanceof Process) { + $process->exit(0); + } + } + + + /** + * @param $tmp + * @return string + */ public static function rename($tmp) { $hash = md5_file($tmp['tmp_name']);