延时队列报错 | php 技术论坛-金年会app官方网

生产者代码

namespace app\amqp\producers;

use hyperf\amqp\annotation\producer;
use hyperf\amqp\message\producerdelayedmessagetrait;
use hyperf\amqp\message\producermessage;
use hyperf\amqp\message\type;

#[producer]
class delaydirectproducer extends producermessage
{
use producerdelayedmessagetrait;

// 设置交换机
protected string $exchange = ‘ext.hyperf.delay’;

// 设置交换机类型
protected type|string $type = type::direct;

// 设置路由键
protected array|string $routingkey = ‘delay-routing-key’; // 这里可以根据需求设置路由键

// 构造函数，接收数据和延迟时间
public function __construct($data, int $delaytime = 5000) // 默认延迟时间为 5000 毫秒
{
$this->payload = $data;
$this->applicationheaders = [‘x-delay’ => $delaytime]; // 设置延迟时间
}

// 你也可以在这里添加日志记录
public function getpayload(): string
{
// log::info(“sending delayed message: “ . json_encode($this->payload));
return $this->payload;
}
}
消费者代码

namespace app\amqp\consumer;

use hyperf\amqp\annotation\consumer;
use hyperf\amqp\message\consumerdelayedmessagetrait;
use hyperf\amqp\message\consumermessage;
use hyperf\amqp\message\type;
use hyperf\amqp\result;
use phpamqplib\message\amqpmessage;

#[consumer(nums: 10)]
class delaydirectconsumer extends consumermessage
{
use consumerdelayedmessagetrait;

protected string $exchange = ‘ext.hyperf.delay’;

protected ?string $queue = ‘queue.hyperf.delay’;

protected type|string $type = type::direct; // 或者 type::fanout
protected array|string $routingkey = ‘’;

public function consumemessage($data, amqpmessage $message): result
{
var_dump($data, ‘delay direct consumetime:’ . (microtime(true)));
return result::ack;
}
}

本作品采用《cc 协议》，转载必须注明作者和本文链接