notifyConsumer.php 6.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175
  1. <?php
  2. /**
  3. * 消息通知消费者
  4. * 处理各种消息通知:新订单、新客户、充值、销账、配送状态变更等
  5. */
  6. namespace common\components\rabbitmq;
  7. use biz\wx\classes\WxMessageClass;
  8. use bizGhs\order\classes\OrderClass;
  9. use bizHd\device\classes\HdDeviceClass;
  10. use bizHd\purchase\classes\PurchaseClass;
  11. use bizHd\shop\classes\ShopClass;
  12. use common\components\noticeUtil;
  13. use common\components\push;
  14. use mikemadisonweb\rabbitmq\components\ConsumerInterface;
  15. use PhpAmqpLib\Message\AMQPMessage;
  16. class notifyConsumer implements ConsumerInterface
  17. {
  18. /**
  19. * 执行消费者逻辑
  20. *
  21. * @param AMQPMessage $msg 消息对象
  22. * @return string 消息处理结果
  23. *
  24. * ConsumerInterface::MSG_ACK - 确认消息(标记为已处理)并从队列中删除
  25. * ConsumerInterface::MSG_REJECT - 拒绝并从队列中删除消息
  26. * ConsumerInterface::MSG_REQUEUE - 拒绝并重新入队消息
  27. */
  28. public function execute(AMQPMessage $msg)
  29. {
  30. try {
  31. // 反序列化消息体
  32. $data = unserialize($msg->body);
  33. if (!is_array($data)) {
  34. noticeUtil::push("通知的消费者报错:Invalid notify message format: {$msg->body}", '15280215347');
  35. return ConsumerInterface::MSG_REJECT;
  36. }
  37. print_r($data);
  38. // 根据通知类型分发处理
  39. $type = $data['type'] ?? null;
  40. switch ($type) {
  41. case 'ghs_new_order_notify':
  42. //供货商的新订单通知
  43. $result = $this->ghsNewOrderNotify($data);
  44. break;
  45. case 'hd_new_order_notify':
  46. //花店的新订单通知
  47. $result = $this->hdNewOrderNotify($data);
  48. break;
  49. case 'hd_new_cg_notify':
  50. //花店的新采购单通知
  51. $result = $this->hdNewCgNotify($data);
  52. break;
  53. case 'ghs_pt_error':
  54. //供货商跑腿异常通知
  55. $result = $this->ghsPtErrorAction($data);
  56. break;
  57. case 'hd_pt_error':
  58. //花店跑腿异常通知
  59. $result = $this->hdPtErrorAction($data);
  60. break;
  61. default:
  62. noticeUtil::push("通知的消费者提示:Unknown notify type: {$type}", '15280215347');
  63. $result = false;
  64. }
  65. if ($result) {
  66. return ConsumerInterface::MSG_ACK;
  67. } else {
  68. noticeUtil::push("通知的消费者提示:Notify message processing failed", '15280215347');
  69. return ConsumerInterface::MSG_REQUEUE;
  70. }
  71. } catch (\Exception $e) {
  72. noticeUtil::push("Notify consumer exception: " . $e->getMessage(), '15280215347');
  73. return ConsumerInterface::MSG_REQUEUE;
  74. }
  75. }
  76. private function ghsPtErrorAction($data)
  77. {
  78. $orderId = $data['orderId'] ?? 0;
  79. $event = $data['event'] ?? '';
  80. $order = OrderClass::getById($orderId, true);
  81. if (!empty($order)) {
  82. $sendNum = $order->sendNum ?? '';
  83. $result = $event . ',取货号 ' . $sendNum;
  84. $customName = $order->customName ?? '';
  85. $shopId = $order->shopId ?? '';
  86. $shop = ShopClass::getById($shopId, true);
  87. if (!empty($shop)) {
  88. WxMessageClass::expressErrorInform($shop, $customName, $orderId, $result);
  89. }
  90. }
  91. return true;
  92. }
  93. private function hdPtErrorAction($data)
  94. {
  95. $orderId = $data['orderId'] ?? 0;
  96. $event = $data['event'] ?? '';
  97. $order = \bizHd\order\classes\OrderClass::getById($orderId, true);
  98. if (!empty($order)) {
  99. $sendNum = $order->sendNum ?? '';
  100. $result = $event . ',取货号 ' . $sendNum;
  101. $customName = $order->customName ?? '';
  102. $shopId = $order->shopId ?? '';
  103. $shop = ShopClass::getById($shopId, true);
  104. if (!empty($shop)) {
  105. WxMessageClass::expressErrorInform($shop, $customName, $orderId, $result);
  106. }
  107. }
  108. return true;
  109. }
  110. private function ghsNewOrderNotify($data)
  111. {
  112. $orderId = $data['orderId'] ?? 0;
  113. if (empty($orderId)) {
  114. return false;
  115. }
  116. $order = OrderClass::getById($orderId, true);
  117. if (empty($order)) {
  118. return false;
  119. }
  120. $shopId = $order->shopId ?? 0;
  121. $shop = ShopClass::getById($shopId, true);
  122. if (empty($shop)) {
  123. return false;
  124. }
  125. if ($shop->actTime != strtotime(date('Y-m-d'))) {
  126. // actTime 活跃时间,只记录以天为单位的时候戳,用于花店注册时推荐批发店
  127. $shop->actTime = strtotime(date('Y-m-d'));
  128. $shop->save();
  129. }
  130. WxMessageClass::ghsHasNewOrderInform($shop, $order);
  131. // 向供货商App发通知
  132. $cgId = $order->purchaseId ?? 0;
  133. $cg = PurchaseClass::getById($cgId, true);
  134. $cgShop = ShopClass::getById($cg->shopId, true, 'shopName, merchantName');
  135. $push = new push('ghs', push::MSG_TYPE_ORDER);
  136. $push->ghsOrderMessage($shop, $cgShop, $order);
  137. return true;
  138. }
  139. private function hdNewOrderNotify($data)
  140. {
  141. return true;
  142. }
  143. private function hdNewCgNotify($data)
  144. {
  145. $cgId = $data['cgId'] ?? 0;
  146. $cg = PurchaseClass::getById($cgId, true);
  147. $shopId = $cg->shopId ?? 0;
  148. $shop = ShopClass::getById($shopId, true);
  149. $allDevices = HdDeviceClass::getAllByCondition(['shopId' => $shopId, 'status' => 1]);
  150. $cids = array_column($allDevices, 'clientId');
  151. $title = '买花成功';
  152. $shopName = $shop->merchantName . ($shop->shopName != '首店' ? '(' . $shop->shopName . ')' : '');
  153. $content = $shopName . ' ¥' . $cg->actPrice;
  154. $payload = [
  155. "page" => "pagesPurchase/purDetails",
  156. "params" => ["id" => $cgId]
  157. ];
  158. $push = new push('hd', push::MSG_TYPE_ORDER);
  159. $push->pushByCloud($cids, $title, $content, $payload);
  160. //$str = implode(',', $cids);
  161. //noticeUtil::push("需要通知:" . $content . ' ' . $str, '15280215347');
  162. return true;
  163. }
  164. }