Queue使用

定义一个队列

  1. namespace App\Utility;
  4. use EasySwoole\Component\Singleton;
  5. use EasySwoole\Queue\Queue;
  7. class MyQueue extends Queue
  8. {
  9.     use Singleton;
  10. }

定义消费进程

  1. namespace App\Utility;
  4. use EasySwoole\Component\Process\AbstractProcess;
  5. use EasySwoole\Queue\Job;
  7. class QueueProcess extends AbstractProcess
  8. {
  10.     protected function run($arg)
  11.     {
  12.         go(function (){
  13.             MyQueue::getInstance()->consumer()->listen(function (Job $job){
  14.                 var_dump($job->toArray());
  15.             });
  16.         });
  17.     }
  18. }

可以多进程，多协程消费

驱动注册

  1. namespace EasySwoole\EasySwoole;
  4. use App\Utility\MyQueue;
  5. use App\Utility\QueueProcess;
  6. use EasySwoole\Component\Timer;
  7. use EasySwoole\EasySwoole\Swoole\EventRegister;
  8. use EasySwoole\EasySwoole\AbstractInterface\Event;
  9. use EasySwoole\Http\Request;
  10. use EasySwoole\Http\Response;
  11. use EasySwoole\Queue\Driver\Redis;
  12. use EasySwoole\Queue\Job;
  13. use EasySwoole\Redis\Config\RedisConfig;
  14. use EasySwoole\RedisPool\RedisPool;
  15. use EasySwoole\Utility\Time;
  18. class EasySwooleEvent implements Event
  19. {
  21.     public static function initialize()
  22.     {
  23.         // TODO: Implement initialize() method.
  24.         date_default_timezone_set('Asia/Shanghai');
  25.     }
  27.     public static function mainServerCreate(EventRegister $register)
  28.     {
  29.         //redis pool使用请看redis 章节文档
  30.         $config = new RedisConfig([
  31.             'host'=>'127.0.0.1'
  32.         ]);
  33.         $redis = new RedisPool($config);
  34.         $driver = new Redis($redis);
  35.         MyQueue::getInstance($driver);
  36.         //注册一个消费进程
  37.         ServerManager::getInstance()->addProcess(new QueueProcess());
  38.         //模拟生产者，可以在任意位置投递
  39.         $register->add($register::onWorkerStart,function ($ser,$id){
  40.             if($id == 0){
  41.                 Timer::getInstance()->loop(3000,function (){
  42.                    $job = new Job();
  43.                    $job->setJobData(['time'=>\time()]);
  44.                    MyQueue::getInstance()->producer()->push($job);
  45.                 });
  46.             }
  47.         });
  49.     }
  51.     public static function onRequest(Request $request, Response $response): bool
  52.     {
  53.         // TODO: Implement onRequest() method.
  54.         return true;
  55.     }
  57.     public static function afterRequest(Request $request, Response $response): void
  58.     {
  59.         // TODO: Implement afterAction() method.
  60.     }
  61. }