延时队列

Delayproducer.Php

Amqpbuilder.Php

AmqpBuilder.php

1. <?php 
2.  
3. declare(strict_types = 1); 
4.  
5. namespace App\Components\Amqp; 
6.  
7. use Hyperf\Amqp\Builder\Builder; 
8.  
9. use Hyperf\Amqp\Builder\QueueBuilder; 
10.  
11. class AmqpBuilder extends QueueBuilder 
12.  
13. { 
14.  
15.     /** 
16.  
17.      * @param array|\PhpAmqpLib\Wire\AMQPTable $arguments 
18.  
19.      * 
20.  
21.      * @return \Hyperf\Amqp\Builder\Builder 
22.  
23.      */ 
24.  
25.     public function setArguments($arguments) : Builder 
26.  
27.     { 
28.  
29.         $this->arguments = array_merge($this->arguments, $arguments); 
30.  
31.         return $this; 
32.  
33.     } 
34.  
35.     /** 
36.  
37.      * 设置延时队列相关参数 
38.  
39.      * 
40.  
41.      * @param string $queueName 
42.  
43.      * @param int    $xMessageTtl 
44.  
45.      * @param string $xDeadLetterExchange 
46.  
47.      * @param string $xDeadLetterRoutingKey 
48.  
49.      * 
50.  
51.      * @return $this 
52.  
53.      */ 
54.  
55.     public function setDelayedQueue(string $queueName, int $xMessageTtl, string $xDeadLetterExchange, string $xDeadLetterRoutingKey) : self 
56.  
57.     { 
58.  
59.         $this->setArguments([ 
60.  
61.             'x-message-ttl'             => ['I', $xMessageTtl * 1000], // 毫秒 
62.  
63.             'x-dead-letter-exchange'    => ['S', $xDeadLetterExchange], 
64.  
65.             'x-dead-letter-routing-key' => ['S', $xDeadLetterRoutingKey], 
66.  
67.         ]); 
68.  
69.         $this->setQueue($queueName); 
70.  
71.         return $this; 
72.  
73.     } 
74.  
75. } 

DelayProducer.php

1. <?php 
2.  
3. declare(strict_types = 1); 
4.  
5. namespace App\Components\Amqp; 
6.  
7. use Hyperf\Amqp\Annotation\Producer; 
8.  
9. use Hyperf\Amqp\Builder; 
10.  
11. use Hyperf\Amqp\Message\ProducerMessageInterface; 
12.  
13. use Hyperf\Di\Annotation\AnnotationCollector; 
14.  
15. use PhpAmqpLib\Message\AMQPMessage; 
16.  
17. use Throwable; 
18.  
19. class DelayProducer extends Builder 
20.  
21. { 
22.  
23.     /** 
24.  
25.      * @param ProducerMessageInterface $producerMessage 
26.  
27.      * @param AmqpBuilder              $queueBuilder 
28.  
29.      * @param bool                     $confirm 
30.  
31.      * @param int                      $timeout 
32.  
33.      * 
34.  
35.      * @return bool 
36.  
37.      * @throws \Throwable 
38.  
39.      */ 
40.  
41.     public function produce(ProducerMessageInterface $producerMessage, AmqpBuilder $queueBuilder, bool $confirm = false, int $timeout = 5) : bool 
42.  
43.     { 
44.  
45.         return retry(1, function () use ($producerMessage, $queueBuilder, $confirm, $timeout) 
46.  
47.         { 
48.  
49.             return $this->produceMessage($producerMessage, $queueBuilder, $confirm, $timeout); 
50.  
51.         }); 
52.  
53.     } 
54.  
55.     /** 
56.  
57.      * @param ProducerMessageInterface $producerMessage 
58.  
59.      * @param AmqpBuilder              $queueBuilder 
60.  
61.      * @param bool                     $confirm 
62.  
63.      * @param int                      $timeout 
64.  
65.      * 
66.  
67.      * @return bool 
68.  
69.      * @throws \Throwable 
70.  
71.      */ 
72.  
73.     private function produceMessage(ProducerMessageInterface $producerMessage, AmqpBuilder $queueBuilder, bool $confirm = false, int $timeout = 5) : bool 
74.  
75.     { 
76.  
77.         $result = false; 
78.  
79.         $this->injectMessageProperty($producerMessage); 
80.  
81.         $message = new AMQPMessage($producerMessage->payload(), $producerMessage->getProperties()); 
82.  
83.         $pool    = $this->getConnectionPool($producerMessage->getPoolName()); 
84.  
85.         /** @var \Hyperf\Amqp\Connection $connection */ 
86.  
87.         $connection = $pool->get(); 
88.  
89.         if ($confirm) { 
90.  
91.             $channel = $connection->getConfirmChannel(); 
92.  
93.         } else { 
94.  
95.             $channel = $connection->getChannel(); 
96.  
97.         } 
98.  
99.         $channel->set_ack_handler(function () use (&$result) 
100.  
101.         { 
102.  
103.             $result = true; 
104.  
105.         }); 
106.  
107.         try { 
108.  
109.             // 处理延时队列 
110.  
111.             $exchangeBuilder = $producerMessage->getExchangeBuilder(); 
112.  
113.             // 队列定义 
114.  
115.             $channel->queue_declare($queueBuilder->getQueue(), $queueBuilder->isPassive(), $queueBuilder->isDurable(), $queueBuilder->isExclusive(), $queueBuilder->isAutoDelete(), $queueBuilder->isNowait(), $queueBuilder->getArguments(), $queueBuilder->getTicket()); 
116.  
117.             // 路由定义 
118.  
119.             $channel->exchange_declare($exchangeBuilder->getExchange(), $exchangeBuilder->getType(), $exchangeBuilder->isPassive(), $exchangeBuilder->isDurable(), $exchangeBuilder->isAutoDelete(), $exchangeBuilder->isInternal(), $exchangeBuilder->isNowait(), $exchangeBuilder->getArguments(), $exchangeBuilder->getTicket()); 
120.  
121.             // 队列绑定 
122.  
123.             $channel->queue_bind($queueBuilder->getQueue(), $producerMessage->getExchange(), $producerMessage->getRoutingKey()); 
124.  
125.             // 消息发送 
126.  
127.             $channel->basic_publish($message, $producerMessage->getExchange(), $producerMessage->getRoutingKey()); 
128.  
129.             $channel->wait_for_pending_acks_returns($timeout); 
130.  
131.         } catch (Throwable $exception) { 
132.  
133.             // Reconnect the connection before release. 
134.  
135.             $connection->reconnect(); 
136.  
137.             throw $exception; 
138.  
139.         } 
140.  
141.         finally { 
142.  
143.             $connection->release(); 
144.  
145.         } 
146.  
147.         return $confirm ? $result : true; 
148.  
149.     } 
150.  
151.     /** 
152.  
153.      * @param ProducerMessageInterface $producerMessage 
154.  
155.      */ 
156.  
157.     private function injectMessageProperty(ProducerMessageInterface $producerMessage) : void 
158.  
159.     { 
160.  
161.         if (class_exists(AnnotationCollector::class)) { 
162.  
163.             /** @var \Hyperf\Amqp\Annotation\Producer $annotation */ 
164.  
165.             $annotation = AnnotationCollector::getClassAnnotation(get_class($producerMessage), Producer::class); 
166.  
167.             if ($annotation) { 
168.  
169.                 $annotation->routingKey && $producerMessage->setRoutingKey($annotation->routingKey); 
170.  
171.                 $annotation->exchange && $producerMessage->setExchange($annotation->exchange); 
172.  
173.             } 
174.  
175.         } 
176.  
177.     } 
178.  
179. } 

处理超时订单

Orderqueueconsumer.Php

Orderqueueproducer.Php

Orderqueueproducer.php

1. <?php 
2.  
3. declare(strict_types = 1); 
4.  
5. namespace App\Amqp\Producer; 
6.  
7. use Hyperf\Amqp\Annotation\Producer; 
8.  
9. use Hyperf\Amqp\Builder\ExchangeBuilder; 
10.  
11. use Hyperf\Amqp\Message\ProducerMessage; 
12.  
13. /** 
14.  
15.  * @Producer(exchange="order_exchange", routingKey="order_exchange") 
16.  
17.  */ 
18.  
19. class OrderQueueProducer extends ProducerMessage 
20.  
21. { 
22.  
23.     public function __construct($data) 
24.  
25.     { 
26.  
27.         $this->payload = $data; 
28.  
29.     } 
30.  
31.     public function getExchangeBuilder() : ExchangeBuilder 
32.  
33.     { 
34.  
35.         return parent::getExchangeBuilder(); // TODO: Change the autogenerated stub 
36.  
37.     } 
38.  
39. } 

Orderqueueconsumer.php

1. <?php 
2.  
3. declare(strict_types = 1); 
4.  
5. namespace App\Amqp\Consumer; 
6.  
7. use App\Service\CityTransport\OrderService; 
8.  
9. use Hyperf\Amqp\Result; 
10.  
11. use Hyperf\Amqp\Annotation\Consumer; 
12.  
13. use Hyperf\Amqp\Message\ConsumerMessage; 
14.  
15. /** 
16.  
17.  * @Consumer(exchange="delay_exchange", routingKey="delay_route", queue="delay_queue", name ="OrderQueueConsumer", nums=1) 
18.  
19.  */ 
20.  
21. class OrderQueueConsumer extends ConsumerMessage 
22.  
23. { 
24.  
25.     public function consume($data) : string 
26.  
27.     { 
28.  
29.        ##业务处理 
30.  
31.     } 
32.  
33.     public function isEnable() : bool 
34.  
35.     { 
36.  
37.         return true; 
38.  
39.     } 
40.  
41. } 

Demo

1. $builder = new AmqpBuilder(); 
2.  
3.         $builder->setDelayedQueue('order_exchange', 1, 'delay_exchange', 'delay_route'); 
4.  
5.         $que = ApplicationContext::getContainer()->get(DelayProducer::class); 
6.  
7.         var_dump($que->produce(new OrderQueueProducer(['order_sn' => (string)mt_rand(10000, 90000)]), $builder))

Tags: Hyperf PHP延时队列

分享到：