hyperf 实现延时队列,处理超时未支付订单

  • Delayproducer.Php
  • Amqpbuilder.Php

AmqpBuilder.php

<?php
declare(strict_types = 1);
namespace App\Components\Amqp;

use Hyperf\Amqp\Builder\Builder;
use Hyperf\Amqp\Builder\QueueBuilder;

class AmqpBuilder extends QueueBuilder
{
    /**
     * @param array|\PhpAmqpLib\Wire\AMQPTable $arguments
     *
     * @return \Hyperf\Amqp\Builder\Builder
     */
    public function setArguments($arguments) : Builder
    {
        $this->arguments = array_merge($this->arguments, $arguments);
        return $this;
    }

    /**
     * 设置延时队列相关参数
     *
     * @param string $queueName
     * @param int    $xMessageTtl
     * @param string $xDeadLetterExchange
     * @param string $xDeadLetterRoutingKey
     *
     * @return $this
     */
    public function setDelayedQueue(string $queueName, int $xMessageTtl, string $xDeadLetterExchange, string $xDeadLetterRoutingKey) : self
    {
        $this->setArguments([
            'x-message-ttl'             => ['I', $xMessageTtl * 1000], // 毫秒
            'x-dead-letter-exchange'    => ['S', $xDeadLetterExchange],
            'x-dead-letter-routing-key' => ['S', $xDeadLetterRoutingKey],
        ]);
        $this->setQueue($queueName);
        return $this;
    }
}

DelayProducer.php

<?php
declare(strict_types = 1);
namespace App\Components\Amqp;

use Hyperf\Amqp\Annotation\Producer;
use Hyperf\Amqp\Builder;
use Hyperf\Amqp\Message\ProducerMessageInterface;
use Hyperf\Di\Annotation\AnnotationCollector;
use PhpAmqpLib\Message\AMQPMessage;
use Throwable;

class DelayProducer extends Builder
{
    /**
     * @param ProducerMessageInterface $producerMessage
     * @param AmqpBuilder              $queueBuilder
     * @param bool                     $confirm
     * @param int                      $timeout
     *
     * @return bool
     * @throws \Throwable
     */
    public function produce(ProducerMessageInterface $producerMessage, AmqpBuilder $queueBuilder, bool $confirm = false, int $timeout = 5) : bool
    {
        return retry(1, function () use ($producerMessage, $queueBuilder, $confirm, $timeout)
        {
            return $this->produceMessage($producerMessage, $queueBuilder, $confirm, $timeout);
        });
    }

    /**
     * @param ProducerMessageInterface $producerMessage
     * @param AmqpBuilder              $queueBuilder
     * @param bool                     $confirm
     * @param int                      $timeout
     *
     * @return bool
     * @throws \Throwable
     */
    private function produceMessage(ProducerMessageInterface $producerMessage, AmqpBuilder $queueBuilder, bool $confirm = false, int $timeout = 5) : bool
    {
        $result = false;

        $this->injectMessageProperty($producerMessage);

        $message = new AMQPMessage($producerMessage->payload(), $producerMessage->getProperties());
        $pool    = $this->getConnectionPool($producerMessage->getPoolName());
        /** @var \Hyperf\Amqp\Connection $connection */
        $connection = $pool->get();
        if ($confirm) {
            $channel = $connection->getConfirmChannel();
        } else {
            $channel = $connection->getChannel();
        }
        $channel->set_ack_handler(function () use (&$result)
        {
            $result = true;
        });

        try {
            // 处理延时队列
            $exchangeBuilder = $producerMessage->getExchangeBuilder();

            // 队列定义
            $channel->queue_declare($queueBuilder->getQueue(), $queueBuilder->isPassive(), $queueBuilder->isDurable(), $queueBuilder->isExclusive(), $queueBuilder->isAutoDelete(), $queueBuilder->isNowait(), $queueBuilder->getArguments(), $queueBuilder->getTicket());

            // 路由定义
            $channel->exchange_declare($exchangeBuilder->getExchange(), $exchangeBuilder->getType(), $exchangeBuilder->isPassive(), $exchangeBuilder->isDurable(), $exchangeBuilder->isAutoDelete(), $exchangeBuilder->isInternal(), $exchangeBuilder->isNowait(), $exchangeBuilder->getArguments(), $exchangeBuilder->getTicket());

            // 队列绑定
            $channel->queue_bind($queueBuilder->getQueue(), $producerMessage->getExchange(), $producerMessage->getRoutingKey());

            // 消息发送
            $channel->basic_publish($message, $producerMessage->getExchange(), $producerMessage->getRoutingKey());
            $channel->wait_for_pending_acks_returns($timeout);
        } catch (Throwable $exception) {
            // Reconnect the connection before release.
            $connection->reconnect();
            throw $exception;
        }
        finally {
            $connection->release();
        }

        return $confirm ? $result : true;
    }

    /**
     * @param ProducerMessageInterface $producerMessage
     */
    private function injectMessageProperty(ProducerMessageInterface $producerMessage) : void
    {
        if (class_exists(AnnotationCollector::class)) {
            /** @var \Hyperf\Amqp\Annotation\Producer $annotation */
            $annotation = AnnotationCollector::getClassAnnotation(get_class($producerMessage), Producer::class);
            if ($annotation) {
                $annotation->routingKey && $producerMessage->setRoutingKey($annotation->routingKey);
                $annotation->exchange && $producerMessage->setExchange($annotation->exchange);
            }
        }
    }
}

处理超时订单

  • Orderqueueconsumer.Php
  • Orderqueueproducer.Php

Orderqueueproducer.php

<?php

declare(strict_types = 1);

namespace App\Amqp\Producer;

use Hyperf\Amqp\Annotation\Producer;
use Hyperf\Amqp\Builder\ExchangeBuilder;
use Hyperf\Amqp\Message\ProducerMessage;

/**
 * @Producer(exchange="order_exchange", routingKey="order_exchange")
 */
class OrderQueueProducer extends ProducerMessage
{
    public function __construct($data)
    {
        $this->payload = $data;
    }

    public function getExchangeBuilder() : ExchangeBuilder
    {
        return parent::getExchangeBuilder(); // TODO: Change the autogenerated stub
    }
}

Orderqueueconsumer.php

<?php

declare(strict_types = 1);

namespace App\Amqp\Consumer;

use App\Service\CityTransport\OrderService;
use Hyperf\Amqp\Result;
use Hyperf\Amqp\Annotation\Consumer;
use Hyperf\Amqp\Message\ConsumerMessage;

/**
 * @Consumer(exchange="delay_exchange", routingKey="delay_route", queue="delay_queue", name ="OrderQueueConsumer", nums=1)
 */
class OrderQueueConsumer extends ConsumerMessage
{
    public function consume($data) : string
    {
       ##业务处理
    }

    public function isEnable() : bool
    {
        return true;
    }
}

demo

$builder = new AmqpBuilder();
        $builder->setDelayedQueue('order_exchange', 1, 'delay_exchange', 'delay_route');
        $que = ApplicationContext::getContainer()->get(DelayProducer::class);
        var_dump($que->produce(new OrderQueueProducer(['order_sn' => (string)mt_rand(10000, 90000)]), $builder))
本作品采用《CC 协议》,转载必须注明作者和本文链接
本帖由系统于 3年前 自动加精
CodingHePing
讨论数量: 9

OrderQueueConsumer 消费队列里面要声明direct

3年前 评论
CodingHePing (楼主) 3年前

需要下载 延时插件 rabbitmq_delayed_message_exchange 吗?,有使用死信 实现的方案demo吗

2年前 评论
CodingHePing (楼主) 1年前

hyperf2.2,按照上面的步骤,报错:Call to undefined method App\Components\DelayProducer::getConnectionPool

1年前 评论
13697332484 (作者) 1年前
CodingHePing (楼主) 1年前
momo714 1年前
momo714 1年前

讨论应以学习和精进为目的。请勿发布不友善或者负能量的内容,与人为善,比聪明更重要!