修改成为 x-delayed-message，现在延迟消息走 header x-delay，不再依赖 TTL+DLX 队列模型

biz-ghs/product/classes/ProductClass.php

@@ -30,6 +30,7 @@ use common\components\orderSn;
 use common\components\stringUtil;
 use common\components\util;
 use common\services\xhItemService;
+use PhpAmqpLib\Wire\AMQPTable;
 use Yii;
 
 class ProductClass extends BaseClass
@@ -2673,13 +2674,15 @@ class ProductClass extends BaseClass
             $producer->publish($message, 'limitBuyDelayExchange', 'limitBuyDelayRoute', [
                 'delivery_mode' => 2,
                 'content_type' => 'application/octet-stream',
-                'expiration' => (string)($seconds * 1000),
+                'application_headers' => new AMQPTable([
+                    'x-delay' => intval($seconds * 1000),
+                ]),
             ]);
             Yii::info('限购延迟消息已发送: ' . json_encode([
                 'type' => 'limit_buy_clear',
                 'productId' => intval($productId),
                 'clearAt' => intval($clearAt),
-                'expireMs' => intval($seconds * 1000),
+                'delayMs' => intval($seconds * 1000),
             ], JSON_UNESCAPED_UNICODE), __METHOD__);
         } else {
             Yii::$app->redis->executeCommand('DEL', [$clearMarkKey]);

common/components/rabbitmq/stockConsumer.php

@@ -30,7 +30,7 @@ class stockConsumer implements ConsumerInterface
             // 反序列化消息体
             $data = unserialize($msg->body);
             if (!is_array($data)) {
-                noticeUtil::push("库存的消费者报错：Invalid notify message format: {$msg->body}", '15280215347');
+                noticeUtil::push("库存的消费者报错：Invalid notify message format: {$msg->body}");
                 return ConsumerInterface::MSG_REJECT;
             }
             print_r($data);
@@ -44,11 +44,9 @@ class stockConsumer implements ConsumerInterface
                 case 'add':
                     $result = true;
                     echo '持久化OK---';
-                    print_r($data);
                     break;
                 case 'limit_buy_clear':
                     echo 'limit_buy_clear---';
-                    print_r($data);
                     $result = $this->clearOrderItemLimitBuy($data);
                     break;
                 default:
@@ -56,16 +54,8 @@ class stockConsumer implements ConsumerInterface
                     $result = false;
             }
             if ($result) {
-                if ($type == 'limit_buy_clear') {
-                    Yii::info('限购延迟消息消费成功(ACK): ' . json_encode($data, JSON_UNESCAPED_UNICODE), __METHOD__);
-                    Yii::getLogger()->flush(true);
-                }
                 return ConsumerInterface::MSG_ACK;
             } else {
-                if ($type == 'limit_buy_clear') {
-                    Yii::warning('限购延迟消息消费失败，准备重入队: ' . json_encode($data, JSON_UNESCAPED_UNICODE), __METHOD__);
-                    Yii::getLogger()->flush(true);
-                }
                 noticeUtil::push("库存的消费者报错：Stock message processing failed");
                 //return ConsumerInterface::MSG_REQUEUE;
                 return ConsumerInterface::MSG_ACK;
@@ -87,40 +77,29 @@ class stockConsumer implements ConsumerInterface
     {
         $productId = intval($data['productId'] ?? 0);
         if ($productId <= 0) {
-            noticeUtil::push('取消限购的消费者报错：limit_buy_clear 缺少 productId', '15280215347');
+            noticeUtil::push('取消限购的消费者报错：limit_buy_clear 缺少 productId');
             return true;
         }
 
         $clearAt = intval($data['clearAt'] ?? 0);
         if ($clearAt <= 0) {
-            noticeUtil::push('取消限购的消费者报错：limit_buy_clear 缺少 clearAt', '15280215347');
+            noticeUtil::push('取消限购的消费者报错：limit_buy_clear 缺少 clearAt');
             return true;
         }
 
         // 旧消息直接忽略，避免“先到期的旧消息”提前清空
         if (!ProductClass::checkLimitBuyClearMessage($productId, $clearAt)) {
-            Yii::info('限购延迟消息已忽略(旧消息): ' . json_encode([
-                'productId' => $productId,
-                'clearAt' => $clearAt,
-            ], JSON_UNESCAPED_UNICODE), __METHOD__);
-            Yii::getLogger()->flush(true);
             return true;
         }
 
         $result = ProductClass::clearLimitBuyByProductId($productId);
         if ($result) {
             ProductClass::clearLimitBuyClearMark($productId);
-            Yii::info('限购字段清理完成: ' . json_encode([
-                'productId' => $productId,
-                'clearAt' => $clearAt,
-            ], JSON_UNESCAPED_UNICODE), __METHOD__);
-            Yii::getLogger()->flush(true);
         } else {
-            Yii::warning('限购字段清理失败: ' . json_encode([
+            noticeUtil::push('限购字段清理失败: ' . json_encode([
                 'productId' => $productId,
                 'clearAt' => $clearAt,
-            ], JSON_UNESCAPED_UNICODE), __METHOD__);
-            Yii::getLogger()->flush(true);
+            ], JSON_UNESCAPED_UNICODE));
         }
         return $result;
     }

common/config/rabbitMQ.php

@@ -39,13 +39,11 @@ $rabbitMQ = [
         ],
         [
             'name' => 'limitBuyDelayExchange',
-            'type' => 'direct',
-            'durable' => true,
-        ],
-        [
-            'name' => 'limitBuyExchange',
-            'type' => 'direct',
+            'type' => 'x-delayed-message',
             'durable' => true,
+            'arguments' => new \PhpAmqpLib\Wire\AMQPTable([
+                'x-delayed-type' => 'direct',
+            ]),
         ]
     ],
 
@@ -73,15 +71,6 @@ $rabbitMQ = [
             'passive' => false,
             'durable' => true,
         ],
-        [
-            'name' => 'limitBuyDelayQueue',
-            'passive' => false,
-            'durable' => true,
-            'arguments' => new \PhpAmqpLib\Wire\AMQPTable([
-                'x-dead-letter-exchange' => 'limitBuyExchange',
-                'x-dead-letter-routing-key' => 'limitBuyRoute',
-            ]),
-        ],
         [
             'name' => 'limitBuyQueue',
             'passive' => false,
@@ -115,14 +104,9 @@ $rabbitMQ = [
             'routing_keys' => ['ptRoute'],
         ],
         [
-            'queue' => 'limitBuyDelayQueue',
+            'queue' => 'limitBuyQueue',
             'exchange' => 'limitBuyDelayExchange',
             'routing_keys' => ['limitBuyDelayRoute'],
-        ],
-        [
-            'queue' => 'limitBuyQueue',
-            'exchange' => 'limitBuyExchange',
-            'routing_keys' => ['limitBuyRoute'],
         ]
     ],