add

app/queue/ActivitiesQueue.php

@@ -0,0 +1,20 @@
+<?php
+
+namespace app\queue;
+
+use app\api\model\Orders;
+use app\api\model\UserMoney;
+use app\common\library\DatabaseRoute;
+use think\facade\Log;
+use think\queue\Job;
+
+/**
+ * 奖项领取
+ */
+class ActivitiesQueue extends BaseQueue
+{
+    public function run(Job $job, $data)
+    {
+        Orders::activities($data['userInfo'], $data['sourceUser']);
+    }
+}

app/queue/BaseQueue.php

@@ -0,0 +1,30 @@
+<?php
+
+namespace app\queue;
+
+use think\facade\Log;
+use think\queue\Job;
+
+abstract class BaseQueue
+{
+    public function fire(Job $job, $data)
+    {
+        $start = microtime(true);
+        try {
+            Log::info("消息队列接收到消息，当前队列: ".self::class.", 携带数据: ".json_encode($data));
+            $this->run($job, $data);
+            Log::info("消息队列执行成功：" . static::class);
+            $job->delete();
+        } catch (\Throwable $e) {
+            Log::error("消息队列执行异常：" . $e->getMessage());
+            Log::info($e->getTraceAsString());
+            $job->release(10); // 或 $job->fail()
+        }
+
+        $end = microtime(true);
+        Log::info("消息队列执行完毕, 耗时：" . ($end - $start) . 's');
+    }
+
+    // 子类实现具体逻辑
+    abstract public function run(Job $job, $data);
+}

app/queue/DiscCompensateJob.php

@@ -0,0 +1,26 @@
+<?php
+
+namespace app\queue;
+
+use app\admin\model\DiscSpinning;
+use think\facade\Log;
+use think\queue\Job;
+
+class DiscCompensateJob
+{
+
+    public function fire(Job $job, $data) {
+        try {
+            Log::write('准备处理DiscCompensateJob' . json_encode($data, JSON_UNESCAPED_UNICODE));
+            DiscSpinning::receive1($data);
+            $job->delete(); // 处理成功删除任务
+        } catch (\Exception $e) {
+            if ($job->attempts() < 3) {
+                $job->release(5); // 重试3次，间隔5秒
+            } else {
+                $job->delete();
+                Log::error("大转盘补偿任务最终失败：ID={$data['id']}");
+            }
+        }
+    }
+}

app/queue/DiscReceiveQueue.php

@@ -0,0 +1,50 @@
+<?php
+
+namespace app\queue;
+
+use app\api\model\UserMoney;
+use app\common\library\DatabaseRoute;
+use think\facade\Log;
+use think\queue\Job;
+
+/**
+ * 奖项领取
+ */
+class DiscReceiveQueue extends BaseQueue
+{
+    public function run(Job $job, $data)
+    {
+        $drawsInfo = $data['draws'];
+        if ($drawsInfo['type'] != 2) {
+            return;
+        }
+
+        $user = DatabaseRoute::getDb('tb_user', $drawsInfo['user_id'])->find();
+        if ($user['status'] != 1) {
+            return;
+        }
+
+        $id = DatabaseRoute::getDb('user_money_details', $drawsInfo['user_id'], true)->insert([
+            'user_id' => $drawsInfo['user_id'],
+            'title' => '现金大转盘',
+            'classify' => 5,
+            'type' => 1,
+            'state' => 2,
+            'money' => $drawsInfo['number'],
+            'content' => "现金红包奖励{$drawsInfo['number']}元",
+            'money_type' => 1,
+            'create_time' => getNormalDate(),
+            'source_id' => $drawsInfo['id']
+        ]);
+
+        DatabaseRoute::getDb('disc_spinning_record', $drawsInfo['user_id'], true, true, false)->where([
+            'id' => $drawsInfo['id']
+        ])->update([
+            'target' => 2,
+            'target_id' => $id
+        ]);
+
+        UserMoney::updateAmount($drawsInfo['user_id'], $drawsInfo['number']);
+
+    }
+}

app/queue/UserPushQueue.php

@@ -0,0 +1,24 @@
+<?php
+
+namespace app\queue;
+
+use app\admin\controller\Cash;
+use app\api\model\UserMoney;
+use app\common\library\DatabaseRoute;
+use think\facade\Log;
+use think\queue\Job;
+
+/**
+ * 奖项领取
+ */
+class UserPushQueue extends BaseQueue
+{
+    public function run(Job $job, $data)
+    {
+        $userInfoList = $data['list'];
+        foreach ($userInfoList as $userInfo) {
+            Cash::send($userInfo, $data['title'], $data['content']);
+        }
+
+    }
+}