stockConsumer.php 2.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960
  1. <?php
  2. /**
  3. * 库存管理消费者
  4. * 处理花材库存补充、扣减等库存操作
  5. */
  6. namespace common\components\rabbitmq;
  7. use common\components\noticeUtil;
  8. use mikemadisonweb\rabbitmq\components\ConsumerInterface;
  9. use PhpAmqpLib\Message\AMQPMessage;
  10. use Yii;
  11. class stockConsumer implements ConsumerInterface
  12. {
  13. /**
  14. * 执行消费者逻辑
  15. *
  16. * @param AMQPMessage $msg 消息对象
  17. * @return string 消息处理结果
  18. *
  19. * ConsumerInterface::MSG_ACK - 确认消息(标记为已处理)并从队列中删除
  20. * ConsumerInterface::MSG_REJECT - 拒绝并从队列中删除消息
  21. * ConsumerInterface::MSG_REQUEUE - 拒绝并重新入队消息
  22. */
  23. public function execute(AMQPMessage $msg)
  24. {
  25. try {
  26. // 反序列化消息体
  27. $data = unserialize($msg->body);
  28. if (!is_array($data)) {
  29. noticeUtil::push("库存的消费者报错:Invalid notify message format: {$msg->body}", '15280215347');
  30. return ConsumerInterface::MSG_REJECT;
  31. }
  32. print_r($data);
  33. // 根据操作类型分发处理
  34. $type = $data['type'] ?? null;
  35. switch ($type) {
  36. case 'add':
  37. $result = true;
  38. echo '持久化OK---';
  39. print_r($data);
  40. break;
  41. default:
  42. noticeUtil::push("库存的消费者报错,未知 type: {$type}");
  43. $result = false;
  44. }
  45. if ($result) {
  46. return ConsumerInterface::MSG_ACK;
  47. } else {
  48. noticeUtil::push("库存的消费者报错:Stock message processing failed");
  49. return ConsumerInterface::MSG_REQUEUE;
  50. }
  51. } catch (\Exception $e) {
  52. noticeUtil::push("库存的消费者报错:" . $e->getMessage());
  53. return ConsumerInterface::MSG_REQUEUE;
  54. }
  55. }
  56. }