PHP code example of zoujingli / type-queue

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

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

    

zoujingli / type-queue example snippets

declare(strict_types=1);

use Type\Queue\Job;
use Type\Queue\JobContext;
use Type\Queue\Message;
use Type\Queue\Queue;
use Type\Queue\Registry;
use Type\Queue\Worker;
use Type\Redis\Purpose;
use Type\Redis\RedisConfiguration;
use Type\Redis\RedisManager;
use Type\Runtime\ExecutionScope;

/** 示例任务只输出稳定消息 ID；实际业务效果需在写入目标原子防重。 */
final class ReadmeJob implements Job
{
    /** @param array<string, mixed> $payload 已由消息协议验证的 JSON 数据。 */
    public function handle(JobContext $context, array $payload): void
    {
        $context->assertActive();
        echo $context->message()->id() . "\n";
    }
}

/**
 * 在启动期启用 I/O hook，并在同一协程内装配、执行及关闭队列资源。
 * 示例命名空间只用于投递与消费演示，不提供业务幂等保证。
 */
function main(): void
{
    \Type\Runtime\CoroutineRuntime::enableIo();
    \Type\Runtime\CoroutineRuntime::run(static function (): void {
        $host = getenv('REDIS_HOST');
        $port = filter_var(getenv('REDIS_PORT') === false ? '6379' : getenv('REDIS_PORT'), FILTER_VALIDATE_INT,
            ['options' => ['min_range' => 1, 'max_range' => 65535]]);
        if (!is_int($port)) {
            throw new InvalidArgumentException('REDIS_PORT 必须为有效整数端口');
        }
        $manager = new RedisManager(['default' => new RedisConfiguration($host === false ? '127.0.0.1' : $host, $port)]);
        $scope = new ExecutionScope();
        try {
            $redis = $manager->connection($scope, 'default', Purpose::SCRIPT);
            $queue = new Queue($redis, 'readme-example', 'example', 30000, 100);
            $registry = new Registry();
            $registry->register('readme.echo', 1, static fn (JobContext $context): Job => new ReadmeJob());
            $queue->publish(new Message('readme-' . bin2hex(random_bytes(8)), 'readme.echo', 1, []));
            $worker = new Worker($queue, $registry, 'readme-worker');
            try {
                $worker->run(1);
            } finally {
                $worker->stop();
            }
        } finally {
            try {
                $scope->close();
            } finally {
                $manager->close();
            }
        }
    });
}