新增了处理优惠卷过期队列和command

app/Amqp/Consumer/CouponExpiredDelayDirectConsumer.php

@@ -0,0 +1,103 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Amqp\Consumer;
+
+use App\Amqp\Producer\CouponExpiredDelayDirectProducer;
+use App\Model\Coupon;
+use App\Model\UserCoupon;
+use App\Utils\Log;
+use Hyperf\Amqp\Message\ConsumerDelayedMessageTrait;
+use Hyperf\Amqp\Message\ProducerDelayedMessageTrait;
+use Hyperf\Amqp\Message\Type;
+use Hyperf\Amqp\Producer;
+use Hyperf\Amqp\Result;
+use Hyperf\Amqp\Annotation\Consumer;
+use Hyperf\Amqp\Message\ConsumerMessage;
+use PhpAmqpLib\Message\AMQPMessage;
+
+/**
+ * 处理优惠卷过期
+ */
+#[Consumer(nums: 1)]
+class CouponExpiredDelayDirectConsumer extends ConsumerMessage
+{
+    use ProducerDelayedMessageTrait;
+    use ConsumerDelayedMessageTrait;
+
+    protected string $exchange = 'amqp.delay.direct';
+
+    protected ?string $queue = 'coupon.expired.delay.queue';
+
+    protected string $type = Type::DIRECT; //Type::FANOUT;
+
+    protected string|array $routingKey = 'CouponExpired';
+
+    public function consumeMessage($data, AMQPMessage $message): string
+    {
+        Log::getInstance("queue-CouponExpired")->info("开始：" . json_encode($data, JSON_UNESCAPED_UNICODE));
+
+        // 检测参数
+        if (!isset($data['coupon_id'])){
+            Log::getInstance("queue-CouponExpired")->error("入参错误");
+            return Result::DROP;
+        }
+
+        // 获取优惠卷数据
+        $params = array();
+        $params['coupon_id'] = $data['coupon_id'];
+        $coupon = Coupon::getOne($params);
+        if (empty($coupon)){
+            Log::getInstance("queue-CouponExpired")->info("错误优惠卷，无需处理");
+            return Result::DROP;
+        }
+
+        // 检测优惠卷状态
+        if ($coupon['coupon_status'] != 1){
+            Log::getInstance("queue-CouponExpired")->info("优惠卷非正常状态，无需处理");
+            return Result::DROP;
+        }
+
+        // 检测优惠卷过期时间
+        if ($coupon['valid_type'] == 2){
+            Log::getInstance("queue-CouponExpired")->info("优惠卷类型为相对时效，无需处理");
+            return Result::DROP;
+        }
+
+        $valid_end_time = strtotime($coupon['valid_end_time']);
+
+        try {
+            // 处理未过期事件
+            // 先删除-重新添加队列
+            $time = $valid_end_time - time();
+            if ($time > 0) {
+                $queue_data = array();
+                $queue_data['coupon_id'] = $coupon['coupon_id'];
+                $message = new CouponExpiredDelayDirectProducer($queue_data);
+                $message->setDelayMs(1000 * $time);
+                $producer = $this->container->get(Producer::class);
+                $res = $producer->produce($message);
+                if (!$res) {
+                    Log::getInstance("queue-CouponExpired")->error("未到过期时间，重新添加队列失败");
+                    return Result::REQUEUE;
+                }
+
+                return Result::DROP;
+            }
+
+            // 修改优惠卷
+            $params = array();
+            $params['coupon_id'] = $coupon['coupon_id'];
+
+            $data = array();
+            $data['coupon_status'] = 3;
+            Coupon::edit($params, $data);
+        }catch (\Throwable $e){
+            Log::getInstance("queue-CouponExpired")->error($e->getMessage());
+            return Result::REQUEUE; // 重回队列
+        }
+
+        return Result::ACK;
+    }
+}

app/Amqp/Consumer/UserCouponExpiredDelayDirectConsumer.php

@@ -77,11 +77,6 @@ class UserCouponExpiredDelayDirectConsumer extends ConsumerMessage
             // 先删除-重新添加队列
             $time = $valid_end_time - time();
             if ($time > 0){
-                $time = $valid_end_time - time();
-                if ($time <= 0){
-                    $time = 60 * 5;
-                }
-
                 $queue_data = array();
                 $queue_data['user_coupon_id'] = $user_coupon['user_coupon_id'];
                 $message = new UserCouponExpiredDelayDirectProducer($queue_data);

app/Amqp/Producer/CouponExpiredDelayDirectProducer.php

@@ -0,0 +1,27 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Amqp\Producer;
+
+use Hyperf\Amqp\Annotation\Producer;
+use Hyperf\Amqp\Message\ProducerDelayedMessageTrait;
+use Hyperf\Amqp\Message\ProducerMessage;
+use Hyperf\Amqp\Message\Type;
+
+#[Producer]
+class CouponExpiredDelayDirectProducer extends ProducerMessage
+{
+    use ProducerDelayedMessageTrait;
+
+    protected string $exchange = 'amqp.delay.direct';
+
+    protected string $type = Type::DIRECT;
+
+    protected string|array $routingKey = 'CouponExpired';
+
+    public function __construct($data)
+    {
+        $this->payload = $data;
+    }
+}

app/Command/AddCouponExpiredQueueCommand.php

@@ -0,0 +1,101 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Command;
+
+use App\Amqp\Producer\CouponExpiredDelayDirectProducer;
+use App\Model\Coupon;
+use App\Model\UserCoupon;
+use Hyperf\Amqp\Producer;
+use Hyperf\Command\Command as HyperfCommand;
+use Hyperf\Command\Annotation\Command;
+use Hyperf\DbConnection\Db;
+use Psr\Container\ContainerInterface;
+
+#[Command]
+class AddCouponExpiredQueueCommand extends HyperfCommand
+{
+    public function __construct(protected ContainerInterface $container)
+    {
+        parent::__construct('AddCouponExpiredQueue');
+    }
+
+    public function configure(): void
+    {
+        parent::configure();
+        $this->setDescription('添加系统即将过期优惠卷添加至优惠卷过期队列');
+    }
+
+    public function handle(): void
+    {
+        $this->line('开始');
+
+        try {
+            // 获取需执行的用户优惠卷
+            $coupons = $this->getExecUserCoupon();
+            if (empty($coupons)){
+                $this->line("结束，无优惠卷可执行");
+                return;
+            }
+        }catch (\Throwable $e){
+            $this->line($e->getMessage());
+            return;
+        }
+
+        foreach ($coupons as $coupon){
+            Db::beginTransaction();
+            try {
+                // 添加优惠卷过期队列
+                $valid_end_time = strtotime($coupon['valid_end_time']);
+
+                $data = array();
+                $data['coupon_id'] = $coupon['coupon_id'];
+
+                $time = $valid_end_time - time();
+                if ($time < 0){
+                    $time = 60 * 5;
+                }
+
+                $message = new CouponExpiredDelayDirectProducer($data);
+                $message->setDelayMs(1000 * $time);
+                $producer = $this->container->get(Producer::class);
+                $res = $producer->produce($message);
+                if (!$res) {
+                    Db::rollBack();
+                    $this->line("添加队列失败");
+                }
+
+                Db::commit();
+            }catch (\Throwable $e){
+                // 修改优惠卷执行失败
+                Db::rollBack();
+                $this->line($e->getMessage());
+            }
+        }
+    }
+
+    /**
+     * 获取需执行的优惠卷
+     * @return array
+     */
+    public function getExecUserCoupon(): array
+    {
+        // 过期使用时间
+        $valid_end_time_1 = date('Y-m-d 00:00:00',time());
+        $valid_end_time_2 = date('Y-m-d 23:59:59',time());
+        $valid_end_time = [$valid_end_time_1,$valid_end_time_2];
+
+        $params = array();
+        $params['coupon_status'] = 1;
+        $params['valid_type'] = 1;
+
+        $coupons = Coupon::getTodayExpiredCoupon($params,$valid_end_time);
+        if (empty($coupons)){
+            return [];
+        }
+
+        return $coupons->toArray();
+    }
+
+}

app/Command/AddUserCouponExpiredQueueCommand.php

@@ -27,7 +27,7 @@ class AddUserCouponExpiredQueueCommand extends HyperfCommand
     public function configure(): void
     {
         parent::configure();
-        $this->setDescription('添加即将过期优惠卷添加至优惠卷过期队列');
+        $this->setDescription('添加用户即将过期优惠卷添加至优惠卷过期队列');
     }
 
     public function handle(): void

app/Model/Coupon.php

@@ -123,4 +123,29 @@ class Coupon extends Model
             ->whereIn("distribution_object",[7])
             ->get();
     }
+
+    /**
+     * 获取今日过期优惠卷
+     * @param array $params
+     * @param array $valid_end_time
+     * @param array $fields
+     * @return Collection|array
+     */
+    public static function getTodayExpiredCoupon(array $params,array $valid_end_time,array $fields = ['*']): Collection|array
+    {
+        return self::where($params)
+            ->whereBetween('valid_end_time', $valid_end_time)
+            ->get($fields);
+    }
+
+    /**
+     * 修改
+     * @param array $params
+     * @param array $data
+     * @return int
+     */
+    public static function edit(array $params = [], array $data = []): int
+    {
+        return self::where($params)->update($data);
+    }
 }