PHP code example of easyswoole / queue

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

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

    

easyswoole / queue example snippets

use Swoole\Coroutine;

Coroutine::set(['hook_flags' => SWOOLE_HOOK_TCP]);

use EasySwoole\Queue\Driver\Redis\Config;
use EasySwoole\Queue\Driver\Redis\Redis as RedisDriver;
use EasySwoole\Queue\Queue;

$config = new Config();
$config->setHost('127.0.0.1');
$config->setPort(6379);
$config->setUsername(null);          // 未启用 ACL 时保持 null
$config->setPassword('password');
$config->setDatabase(0);             // Redis Cluster 只能使用数据库 0
$config->setConnectTimeout(3.0);
$config->setReadTimeout(3.0);

// setMinObjectNum() 应先于较小的 setMaxObjectNum() 调用。
$config->setMinObjectNum(1);
$config->setMaxObjectNum(20);

$queue = new Queue(new RedisDriver($config));

use EasySwoole\Queue\Driver\Redis\Config;
use EasySwoole\Queue\Driver\Redis\Redis as RedisDriver;
use EasySwoole\Queue\Queue;

$config = new Config();
$config->setMode(Config::MODE_CLUSTER);
$config->setSeeds([
    '127.0.0.1:7001',
    '127.0.0.1:7002',
    '127.0.0.1:7003',
]);
$config->setUsername(null);
$config->setPassword('password');
$config->setConnectTimeout(3.0);
$config->setReadTimeout(3.0);
$config->setMinObjectNum(1);
$config->setMaxObjectNum(20);

$queue = new Queue(new RedisDriver($config));

use Swoole\Coroutine;

Coroutine::set([
    'hook_flags' => SWOOLE_HOOK_TCP | SWOOLE_HOOK_STREAM_FUNCTION,
]);

use EasySwoole\Queue\Driver\RabbitMQ\Config;
use EasySwoole\Queue\Driver\RabbitMQ\RabbitMQ;
use EasySwoole\Queue\Queue;

$config = new Config();
$config->setHost('127.0.0.1');
$config->setPort(5672);
$config->setUsername('easyswoole');
$config->setPassword('password');
$config->setVirtualHost('/easyswoole');
$config->setConnectTimeout(3.0);
$config->setReadWriteTimeout(3.0);
$config->setHeartbeat(0);
$config->setMinObjectNum(1);
$config->setMaxObjectNum(20);

$queue = new Queue(new RabbitMQ($config));

$config->setHeartbeat(10);
$config->setReadWriteTimeout(25.0);

use EasySwoole\Queue\Job;

$job = new Job();
$job->setJobData(['orderId' => 1001]);
$queue->producer('order')->push($job);

$job = $queue->consumer('order')->pop(3.0);
if ($job !== null) {
    var_dump($job->getJobData());
}

$job = new Job();
$job->setJobData(['orderId' => 1001]);
$job->setDelayTime(5);
$queue->producer('order')->push($job);

$job = new Job();
$job->setJobData(['orderId' => 1001]);
$job->setRetryTimes(3);
$job->setWaitConfirmTime(10);
$queue->producer('order')->push($job);

$consumer = $queue->consumer('order');
$job = $consumer->pop(3.0);
if ($job !== null) {
    // 业务处理成功后，确认本次 pop 返回的 Job。
    $consumer->confirm($job);
}

$consumer = $queue->consumer('order');
$consumer->setBreakTime(0.1);
$consumer->setOnBreak(function (): void {
    // 当前没有可消费任务。
});

$consumer->listen(
    function (Job $job) use ($consumer): void {
        try {
            // 处理任务……
            if ($job->getRetryTimes() > 0) {
                $consumer->confirm($job);
            }
        } catch (Throwable $throwable) {
            // 不确认可信任务，等待确认期限到达后重新投递。
        }
    },
    [],
    function (Throwable $throwable, ?Job $job): void {
        // 记录消费异常。
    }
);

use EasySwoole\Queue\Driver\Redis\Config;
use EasySwoole\Queue\Driver\Redis\Redis as RedisDriver;
use EasySwoole\Queue\Job;
use EasySwoole\Queue\Queue;
use Swoole\Coroutine;

 $config->setPassword('password');
    $config->setMinObjectNum(1);
    $config->setMaxObjectNum(20);

    $queue = new Queue(new RedisDriver($config));
    $producer = $queue->producer('example');
    $consumer = $queue->consumer('example');

    $job = new Job();
    $job->setJobData('hello queue');
    $producer->push($job);

    $popped = $consumer->pop(1.0);
    if ($popped !== null) {
        var_dump($popped->getJobData());
    }
});