rabbitmq想实现延迟队列,使用的 workerman/rabbitmq扩展

延迟交换机插件启用了嘛?

消费

<?php

declare(strict_types=1);

namespace app\command;

use Bunny\Channel;
use Bunny\Message;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Output\OutputInterface;
use Workerman\Events\Fiber;
use Workerman\Events\Revolt;
use Workerman\RabbitMQ\Client;
use Workerman\Worker;

class DelayReceive extends Command
{
    protected static $defaultName = 'delayReceive';
    protected static $defaultDescription = 'delayReceive';

    /**
     * @return void
     */
    protected function configure()
    {
        $this->addArgument('name', InputArgument::OPTIONAL, 'Name description');
    }

    /**
     * @param InputInterface $input
     * @param OutputInterface $output
     * @return int
     */
    protected function execute(InputInterface $input, OutputInterface $output): int
    {
        $worker = new Worker();
        $worker->eventLoop = Revolt::class;
        $worker->onWorkerStart = function () {
            $exchange_name = 'delayed_demo_exchange';
            $queue_name = 'demo_delay_queue';
            $routing_key = 'delayed_key';
            $client = Client::factory([
                'host' => 'rabbitmq',
                'port' => 5672,
                'user' => 'guest',
                'password' => 'guest',
                'vhost' => '/',
                'heartbeat' => 60,
            ])->connect();
            $channel = $client->channel();

            // 创建延迟队列
            $channel->queueDeclare($queue_name, false, true, false, false, false, [
                [
                    'x-delayed-type' => 'direct',
                ]
            ]);
            // Consumer
            $channel->consume(
                function (Message $message, Channel $channel, \Bunny\AbstractClient $client) {
                    echo " [>] Received ", $message->content, "Now:",date("Y-m-d H:i:s"), "\n";

                },
                $queue_name,
                $routing_key,
                false,
                true,
                false,
                false,
                [
                    'x-delayed-type' => 'direct',
                ]
            );

            $client->run();

            echo ' [*] Waiting for messages. To exit press CTRL+C', "\n";
        };
        Worker::runAll();
        return self::SUCCESS;
    }

}

生产

<?php

namespace app\command;

use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Output\OutputInterface;
use Workerman\RabbitMQ\Client;

class Delay extends Command
{
protected static $defaultName = 'delay';
protected static $defaultDescription = 'delay';

/**
 * @return void
 */
protected function configure()
{
    // $this->addArgument('name', InputArgument::OPTIONAL, 'Name description');
}

/**
 * @param InputInterface $input
 * @param OutputInterface $output
 * @return int
 */
protected function execute(InputInterface $input, OutputInterface $output): int
{
    // $name = $input->getArgument('name');
    // $output->writeln('Hello delay');

    $client = Client::factory([
        'host' => 'rabbitmq',
        'port' => 5672,
        'user' => 'guest',
        'password' => 'guest',
        'vhost' => '/',
        'heartbeat' => 60,
    ])->connect();

    $channel = $client->channel();
    $exchange_name = 'delayed_demo_exchange';
    $queue_name = 'demo_delay_queue';
    $routing_key = 'delayed_key';
    // 创建延迟交换机，交换机类型为 direct
    $channel->exchangeDeclare(
        $exchange_name,
        'x-delayed-message',
        false,
        true,
        false,
        false,
        true,
        [
            'x-delayed-type' => 'direct',
        ]
    );

    // 创建延迟队列
    $channel->queueDeclare($queue_name, false, true, false, false);
    $channel->queueBind($queue_name, $exchange_name, $routing_key);
    while (true) {
        // 发送消息
        $message = 'Delayed Message at ' . time();
        $channel->publish($message, ['delivery-mode' => 2,'x-delay' => 5000], $exchange_name, $routing_key);
        sleep(1); // 1s 循环一次
    }
    return self::SUCCESS;
}

}

写了个demo 你先对照一下