PHP code example of egret / queue

    
        
<?php
require_once('vendor/autoload.php');

/* Start to develop here. Best regards https://php-download.com/ */

    

egret / queue example snippets

"php": "^7.1.3",
"ext-swoole": "^2.0 || ^3.0 || ^4.0",

namespace Egret\Queue\Test;

use Egret\Queue\AbstractQueueCommand;
use Monolog\Handler\RotatingFileHandler;
use Monolog\Logger;
use Egret\Queue\DingTalk;

// 继承这个抽象类
class TestRedisQueue extends AbstractQueueCommand
{
    protected function configure()
    {
        // 调用父类的configure方法，必须要有
        parent::configure();
        // 设置队列名称，必须要有，设置队列描述（非必须）
        $this->setName('redis')->setDescription('测试redis队列');
        // 设置队列驱动，不设置默认redis，目前只支持redis/kafka
        $this->driver = 'redis';
        // 设置redis连接的配置，这里测试时用，相关的配置信息往下看
        $this->config = [
            'host' => '127.0.0.1',
            'port' => 6379,
            'password' => '123456',
            'database' => 10,
            'predis_options' => [
                // 指定缓存的前缀
                'prefix'  => 'queue:test:'
            ]
        ];
        // 设置pid文件的存放路径，可以不设置，不设置不能用命令停止队列，这里配置的是我测试的路径，大家按自己的情况设置值
        $this->pidPath = CONSOLE_PATH . '/test/runtime';
    }

    // 设置Monolog的日志，设置之后会写入任务的相关日志，可不设置
    protected function getLogger()
    {
        $logger = new Logger('queue');
        $rotating = new RotatingFileHandler(CONSOLE_PATH . '/test/runtime/info.log', 30);
        $logger->pushHandler($rotating);
        return $logger;
    }

    // 配置钉钉机器人，任务异常时会报警到钉钉上，可不设置
    protected function getDingTalk()
    {
        $config = [
            'token' => 'xxxx',
            'secret' => 'xxxx',
            'sign' => true
        ];
        return new DingTalk($config['token'], $config['sign'], $config['secret']);
    }
}

namespace Egret\Queue\Test;

use Egret\Queue\AbstractJob;

class TestJob extends AbstractJob
{
    protected $name;
    protected $flag;

    public function __construct($name, $flag = true)
    {
        $this->name = $name;
        $this->flag = $flag;
    }

    public function execute()
    {
        echo sprintf('queue %s is run' . PHP_EOL, $this->name);
        return $this->flag;
    }
}

$job = new TestJob($this->getName(), false);
(new RedisQueue($redisConf, $topic))->produce($job);

$job = new TestJob($this->getName(), false);
$kafkaConf = [
     'global' => [
         'group.id' => 'test-group',
         'metadata.broker.list' => '127.0.0.1:9092',
         'enable.auto.commit' => 'false',
     ],
     'topic' => [
         'auto.offset.reset' => 'latest',
     ],
];
(new KafkaQueue($kafkaConf, $topic))->produce($job);

use Egret\Queue\Test\TestConsumer;
use Egret\Queue\Test\TestCustomErrorQueue;
use Egret\Queue\Test\TestDingTalkQueue;
use Egret\Queue\Test\TestKafkaQueue;
use Egret\Queue\Test\TestLoggerQueue;
use Egret\Queue\Test\TestRedisQueue;

return [
    TestDingTalkQueue::class,
    TestKafkaQueue::class,
    TestLoggerQueue::class,
    TestRedisQueue::class,
    TestConsumer::class,
    TestCustomErrorQueue::class,

];

shell script
./queue redis status

Queue: redis
PID file: /queue-redis.pid, PID: 0

+-----------+-------+-------+------+--------+------------------------------+
| USER      | PID   | RSS   | STAT | START  | COMMAND                      |
+-----------+-------+-------+------+--------+------------------------------+
| zbangtang | 11993 | 6720  | S+   | 3:28PM | php ./queue redis start test |
| zbangtang | 11984 | 16056 | S+   | 3:28PM | php ./queue redis start test |
+-----------+-------+-------+------+--------+------------------------------+

shell script
./queue redis stop

shell script
-- 项目demo  
　　-- app文件夹  
　　　　-- 可在app文件夹下创建console.php  
　　-- application文件夹   
　　　　-- 可在application文件夹下创建console.php  
　　-- src文件夹   
　　　　-- 可在src文件夹下创建console.php  
　　-- vendor  
　　-- 可在这一层目录下创建console.php  
　　-- composer.json  
　　-- composer.lock