PHP 框架 Hyperf 实现处理超时未支付订单和延时队列

谁借莪１个温暖的怀抱￠ 2023-03-13 15:19 19阅读 0赞

**延时队列**

*  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))

更多学习内容请访问：

[腾讯T3-T4标准精品PHP架构师教程目录大全，只要你看完保证薪资上升一个台阶（持续更新）][T3-T4_PHP]

**以上内容希望帮助到大家，**很多PHPer在进阶的时候总会遇到一些问题和瓶颈，业务代码写多了没有方向感，不知道该从那里入手去提升，对此我整理了一些资料，包括但不限于：分布式架构、高可扩展、高性能、高并发、服务器性能调优、TP6，laravel，YII2，Redis，Swoole、Swoft、Kafka、Mysql优化、shell脚本、Docker、微服务、Nginx等多个知识点高级进阶干货需要的可以免费分享给大家**，需要的可以加入**我的官方群[点击此处][Link 1]。

[T3-T4_PHP]: https://zhuanlan.zhihu.com/p/102278033
[Link 1]: https://link.zhihu.com/?target=https%3A//jq.qq.com/%3F_wv%3D1027%26k%3D5kN6y8c