初始化

app/queue/redis/CancelRecharge.php

@@ -0,0 +1,35 @@
+<?php
+
+namespace app\queue\redis;
+
+use addons\webman\model\PlayerRechargeRecord;
+use ExAdmin\ui\traits\queueProgress;
+use think\Exception;
+use Webman\RedisQueue\Consumer;
+
+class CancelRecharge implements Consumer
+{
+    use queueProgress;
+
+    public $queue = 'cancel_recharge';
+
+    public $connection = 'default';
+
+    /**
+     * @param $data
+     * @throws Exception
+     */
+    public function consume($data)
+    {
+        /** @var PlayerRechargeRecord $playerRechargeRecord */
+        $playerRechargeRecord = PlayerRechargeRecord::where('type', PlayerRechargeRecord::TYPE_REGULAR)
+            ->where('status', PlayerRechargeRecord::STATUS_WAIT)
+            ->where('id', $data['id'])
+            ->first();
+        if (!empty($playerRechargeRecord)) {
+            $playerRechargeRecord->status = PlayerRechargeRecord::STATUS_RECHARGED_SYSTEM_CANCEL;
+            $playerRechargeRecord->cancel_time = date('Y-m-d H:i:s');
+            $playerRechargeRecord->save();
+        }
+    }
+}

app/queue/redis/SendBroadcasts.php

@@ -0,0 +1,27 @@
+<?php
+
+namespace app\queue\redis;
+
+use support\Log;
+use Webman\Push\PushException;
+use Webman\RedisQueue\Consumer;
+
+class SendBroadcasts implements Consumer
+{
+    // 要消费的队列名
+    public $queue = 'broadcast_tasks';
+
+    // 连接名，对应 plugin/webman/redis-queue/redis.php 里的连接`
+    public $connection = 'default';
+
+    // 消费
+    public function consume($data)
+    {
+        // 无需反序列化
+        try {
+            sendSocketMessage('broadcast', $data);
+        } catch (PushException $e) {
+            Log::error($e->getMessage());
+        }
+    }
+}