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

延时队列

• Delayproducer.Php

• Amqpbuilder.Php

AmqpBuilder.php

arguments = array_merge($this->arguments, $arguments);
 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

produceMessage($producerMessage, $queueBuilder, $/confirm/i, $timeout);
 });
    }
    
    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());
 
 $connection = $pool->get();
 if ($/confirm/i) {
     $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;
    }
    
    private function injectMessageProperty(ProducerMessageInterface $producerMessage) : void
    {
 if (class_exists(AnnotationCollector::class)) {
     
     $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

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

Orderqueueconsumer.php

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))
推荐教程：《PHP教程》
以上就是PHP 框架 Hyperf 实现处理超时未支付订单和延时队列的详细内容，更多请关注考高分网其它相关文章！