Mix Worker Pool

Installation

composer require mix/worker-pool

单次调度

$jobQueue =newSwoole\Coroutine\Channel(200);$maxWorkers =100;$handler =function ($data) {// do something};$pool =newMix\WorkerPool\WorkerPool($jobQueue,$maxWorkers,$handler);go(function ()use ($jobQueue,$pool) {// 投放任务for ($i =0;$i <1000;$i++) {$jobQueue->push($i);    }// 停止$pool->stop();});$pool->run();// 阻塞等待

class FooHandlerimplements \Mix\WorkerPool\RunInterface{publicfunctiondo($data):void    {// do something    }}$pool =newMix\WorkerPool\WorkerPool($jobQueue,$maxWorkers,newFooHandler());

常驻调度

$maxWorkers =20;$maxQueue =10;$jobQueue =newSwoole\Coroutine\Channel($maxQueue);$handler =function ($data) {// do something};$pool =newMix\WorkerPool\WorkerPool($jobQueue,$maxWorkers,$handler);$quit =newSwoole\Coroutine\Channel();foreach ([SIGHUP,SIGINT,SIGTERM]as$signal) {Swoole\Process::signal($signal,function ()use ($quit) {$quit->push(true);    });}go(function ()use ($jobQueue,$pool,$quit) {// 投放任务while (true) {if (!$quit->isEmpty()) {$pool->stop();return;        }try {$data =$redis->brPop(['test'],1);        }catch (\Throwable$ex) {// print log$pool->stop();return;        }if (!$data) {continue;        }$data =array_pop($data);// brPop命令最后一个键才是值$jobQueue->push($data);    }});$pool->run();// 阻塞等待

异常处理

class FooHandlerimplements \Mix\WorkerPool\RunInterface{publicfunctiondo($data):void    {try {// do something        }catch (\Throwable$ex){// print log        }    }}

License