README

一个用于控制数据库中队列进程的PHP库

简介

该库用于处理队列中的进程，使用数据库表中的队列，可以获取进程的返回值及其状态，可以启动不仅与当前请求线程异步执行的进程，还可以对其进行监控。

安装

composer require molecular/queueprocessmanager dev-master

要求

• PHP >= 5.4.*

基本用法

使用主要包括一个客户端，它将进程放入队列，同样也有消费者，它们从队列中移除项目并对其进行处理

客户端代码

        //É criado o manipulador da fila, nesse caso por sql
        $handle = new \PhpQPM\QueueHandle\Sql\SqlQueueHandle();
        
        //Informações para conexões do banco
        $handle->connect('mysql:host=localhost;dbname=QueueManager', 'teste', 'teste');
        
        //Cria um Manager para servir de interface para a fila, usando o manipulador sql criado anteriormente
        $manager = new \PhpQPM\Manager($handle);
        
        $foo = 10;
        
        //É colocado uma processo na fila para ser executado, nesse caso um closure
        //O scope do closure é armazenado na fila junto ao closure para ser executado mais tarde
        //o método putProcessOnQueue retorna um observer da fila
        $process = $manager->putProcessOnQueue(function() use($foo){
           return 10 + $foo;
        });
        
        //aguarda o processo ser finalizado
        //O método update do observer atualiza os dados do objeto com base nos dados da fila
        //Após o consumidor da consumir o processo, o método isFinish retornará true
        while(!$process->update()->isFinish()){sleep(1);};
        
        //O método é executado e pode ser obtido o retorno através do método getReturn
        $bar = $process->getReturn();// 20

消费者代码

        //É criado a conexão e o Manager da mesma forma que no cliente
        $handle = new \PhpQPM\QueueHandle\Sql\SqlQueueHandle();
        $handle->connect('mysql:host=localhost;dbname=QueueManager', 'teste', 'teste');
        $manager = new \PhpQPM\Manager($handle);
        //Através do manager pode ser gerado o consumidor da fila
        $worker = $manager->createWorker();
        //O método work cria um loop onde é verificado se existe itens pendentes na fila
        //Caso exista itens pendentes na fila, os mesmo são processados
        $worker->work();

简单。 =)

特性

PHP 队列进程管理器使用 Super_Closure 来序列化 Closure 并将其放入队列，支持多种 Closure 形式，并具有序列化作用域的功能。

除了 Closure 之外，还可以扩展 \PhpQPM\Process\ProcessQueueable 类并将具有 run 方法的类放入队列，该方法将在从队列获取时执行，除了 run 方法外，还可以直接访问进程。

ProcessQueueable 示例

        //A Classe deve estender ProcessQueueable para poder ser executada na fila
        class SimpleProcess extends \PhpQPM\Process\ProcessQueueable
        {
            //A classe possui o método abstrato run que deve ser implementado com a lógica do processo
            //A classe também pode conter propriedades que as mesmas são serializados para a fila
            public function run()
            {
                return 10;
            }
        }
        
        $handle = new \PhpQPM\QueueHandle\Sql\SqlQueueHandle();
        $handle->connect('mysql:host=localhost;dbname=QueueManager', 'teste', 'teste');
        $manager = new \PhpQPM\Manager($handle);
        
        //O método putProcessOnQueue do Manager aceita um objeto do tipo ProcessQueueable
        $process = $manager->putProcessOnQueue(new SimpleProcess());
        
        while(!$process->update()->isFinish()){sleep(1);};
        
        $bar = $process->getReturn();// 10

处理进程

将项目放入队列时，会获得一个观察者，可以监控队列中的进程，该观察者具有的一些功能包括检查状态、获取队列的返回值以及更新观察者本身

    /* Mesmo Código do exemplo anterior*/
    //...
        
        $process = $manager->putProcessOnQueue(new SimpleProcess());
        
        //Verifica se o processo já foi reservado pela fila
        $process->isReserved();
        
        //Verifica se o processo já foi reservado pela fila, porém o seu processamento ainda não foi concluido
        $process->isRunning();
        
        //Verifica se o processo já foi concluído
        $process->isFinish();
        
        //Verifica se o processo ainda não foi reservado pela fila
        $process->isWait();
        
        //Obtem o retorno do processo
        $process->getReturn();
        
        //Verifica se ocorreu algum erro no processamento
        $process->hasError();
        
        //Obtem o erro no processamento
        $process->getError();
        
        //Atualiza o observer com os dados mais atuais da fila
        $process->update();
        
    //...