From 7e820ee9c49d07a3143c5d2d730301f9b69c0abc Mon Sep 17 00:00:00 2001 From: mzl Date: Sun, 12 Oct 2025 06:22:29 +0000 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E6=B6=88=E6=81=AF=E6=8E=A8?= =?UTF-8?q?=E9=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/common/service/MenuBadgeService.php | 178 ++++++++++++++++++ .../service/MessageNotificationsService.php | 2 +- app/common/utils/WsEvents.php | 46 +++-- .../approval/service/ApprovalFlowService.php | 163 ++++++++++++++-- app/controller/IndexController.php | 19 +- ...mUpdate.php => InquiryToCrmUpdate_bak.php} | 2 +- ...{InquiryToCrm.php => InquiryToCrm_bak.php} | 2 +- app/queue/redis/WsMsg.php | 67 +++++++ 8 files changed, 441 insertions(+), 38 deletions(-) create mode 100644 app/common/service/MenuBadgeService.php rename app/queue/redis/{InquiryToCrmUpdate.php => InquiryToCrmUpdate_bak.php} (98%) rename app/queue/redis/{InquiryToCrm.php => InquiryToCrm_bak.php} (98%) create mode 100644 app/queue/redis/WsMsg.php diff --git a/app/common/service/MenuBadgeService.php b/app/common/service/MenuBadgeService.php new file mode 100644 index 0000000..e46997b --- /dev/null +++ b/app/common/service/MenuBadgeService.php @@ -0,0 +1,178 @@ +叶），每一级返回“完整url + 角标” + * @param int $menuId 叶子或中间节点ID（directory/menu均可） + * @param int $leafCount 该节点需要向上累加的消息数（正数可表示新增） + * @param bool $includeHidden 是否包含隐藏/禁用菜单（默认 false：过滤 status != 1） + * @return array + */ + public static function buildBadgeChain( + int $menuId, + int $leafCount, + bool $includeHidden = false, + bool $forceIncludeSelf = true + ): array { + // 取到从“叶->根”的节点列表 + $nodes = self::getAncestorNodesIterative($menuId, $includeHidden, $forceIncludeSelf); + if (empty($nodes)) { + return []; + } + + // 调整为“根->叶” + $nodes = array_reverse($nodes); + $segments = []; + $result = []; + $leafId = $menuId; + + foreach ($nodes as $n) { + $isLeaf = ((int)$n['id'] === $leafId); + + // 中间层的 permission 不展示（但若有内部url段，仍需累计到segments） + if (($n['type'] ?? '') === 'permission' && !$isLeaf) { + // 若这是 internal 且有 url 段，仍拼到段里，便于子级正确拼接 + $openType = strtolower((string)($n['open_type'] ?? 'internal')); + if ($openType === 'internal') { + $seg = trim((string)($n['url'] ?? ''), '/'); + if ($seg !== '') { + $segments[] = $seg; + } + } + continue; + } + + // 计算完整URL（internal 按段拼接；external/iframe 直接用 link/url） + $fullUrl = self::composeFullUrl($segments, $n); + + // 叶子（传入ID）必须返回；其他节点照常 + if ($isLeaf || ($n['type'] ?? '') !== 'permission') { + $result[] = [ + 'id' => (int)$n['id'], + 'name' => (string)$n['name'], + 'type' => (string)$n['type'], + 'url' => $fullUrl, + 'badge' => (int)$leafCount, + ]; + } + } + + return $result; + } + + /** + * 批量聚合多个叶子节点的角标，把同一路径处的角标相加（父级=所有子级之和） + * @param array $items + * @param bool $includeHidden + * @return array + */ + public static function aggregateChains(array $items, bool $includeHidden = false): array + { + $acc = []; // key = url + + foreach ($items as $it) { + $menuId = (int)($it['menu_id'] ?? 0); + $leafCount = (int)($it['count'] ?? 0); + if ($menuId <= 0 || $leafCount === 0) { + continue; + } + + $chain = self::buildBadgeChain($menuId, $leafCount, $includeHidden); + foreach ($chain as $node) { + $key = $node['url']; + if (!isset($acc[$key])) { + $acc[$key] = ['url' => $node['url'], 'badge' => 0]; + } + $acc[$key]['badge'] += $node['badge']; + } + } + + // 父级在前（更短的路径在前），便于前端逐级渲染 + usort($acc, function ($a, $b) { + return substr_count($a['url'], '/') <=> substr_count($b['url'], '/'); + }); + + return array_values($acc); + } + + /** + * 迭代法：从叶子一路查到根（MySQL版本无关） + * $forceIncludeSelf=true 时，无论叶子 status 是否为1，都会把叶子纳入； + * 但祖先节点仍按 $includeHidden 规则过滤。 + */ + private static function getAncestorNodesIterative( + int $id, + bool $includeHidden = false, + bool $forceIncludeSelf = true + ): array { + $cols = ['id', 'parent_id', 'name', 'type', 'url', 'open_type', 'link', 'is_hide', 'status']; + $nodes = []; + $seen = []; + $cur = $id; + $maxHop = 32; // 防环 + + $isFirst = true; + for ($i = 0; $i < $maxHop && $cur > 0; $i++) { + /** @var object|null $row */ + $row = Db::table('resources')->select($cols)->where('id', $cur)->first(); + if (!$row) break; + + $rowArr = (array)$row; + + if (!$includeHidden) { + // 叶子节点：若 $forceIncludeSelf=true 则无视 status 过滤；祖先照常过滤 + if (!$isFirst && (int)$rowArr['status'] !== 1) { + break; + } + if ($isFirst && !$forceIncludeSelf && (int)$rowArr['status'] !== 1) { + // 如果明确不强制包含叶子，叶子也按规则过滤 + break; + } + } + + $nodes[] = $rowArr; + + if (in_array($cur, $seen, true)) { + break; // 防环 + } + $seen[] = $cur; + + $cur = (int)$rowArr['parent_id']; + $isFirst = false; + } + + return $nodes; // 叶 -> 根 + } + + /** + * 组装“完整路由” + * - internal：把父级段 + 自身 url 段拼成形如 /a/b/c + * - external/iframe：直接用 link（若无则退回自身 url） + * - 忽略空段（避免出现重复斜杠） + */ + private static function composeFullUrl(array &$segments, array $node): string + { + $openType = strtolower((string)($node['open_type'] ?? 'internal')); + + // 外链或 iframe：直接返回 link/url，不再参与分段拼接 + if ($openType !== 'internal') { + $target = trim((string)($node['link'] ?? ''), " \t\r\n"); + if ($target === '') { + $target = (string)($node['url'] ?? ''); + } + return $target; + } + + // internal：把自身 url 作为一个“段”拼到前面来的 segments + $seg = trim((string)($node['url'] ?? ''), '/'); + if ($seg !== '') { + $segments[] = $seg; + } + + return '/' . implode('/', $segments); + } +} \ No newline at end of file diff --git a/app/common/service/MessageNotificationsService.php b/app/common/service/MessageNotificationsService.php index cebd4b9..3d49a6a 100644 --- a/app/common/service/MessageNotificationsService.php +++ b/app/common/service/MessageNotificationsService.php @@ -154,7 +154,7 @@ class MessageNotificationsService ]; // 执行推送 - (new \app\common\utils\WsEvents())->pushWsNotice($data, $userId); + (new \app\common\utils\WsEvents())->pushWebsocket('notify.receive', $userId, $data); } return true; } diff --git a/app/common/utils/WsEvents.php b/app/common/utils/WsEvents.php index a0f9a4c..64ff97b 100644 --- a/app/common/utils/WsEvents.php +++ b/app/common/utils/WsEvents.php @@ -53,6 +53,16 @@ class WsEvents } } + protected function eventToMethod(string $event): string + { + $parts = explode('.', $event); + $method = array_shift($parts); + foreach ($parts as $part) { + $method .= ucfirst($part); + } + return $method; + } + // ================= 系统事件 ================= /** @@ -112,6 +122,14 @@ class WsEvents 'system.auth', ['uid' => $user->uid] )); + + // 初始化推送 + $wdata = [ + 'is_first' => true, + 'event' => 'approval.todo.counts', + 'user_id' => $user->uid + ]; + \Webman\RedisQueue\Redis::send('ws-msg', $wdata); } catch (\Throwable $th) { flg('[x] [TOKEN认证] TOKEN认证失败：'.$th->getMessage(), false); Gateway::sendToClient($client_id, WsResponse::error( @@ -235,22 +253,20 @@ class WsEvents )); } - /** - * 推送消息通知 - */ - public function pushWsNotice(array $wsData, int $userId) - { - Gateway::sendToUid($userId, WsResponse::success('notify.receive', $wsData)); - } - // ================= 辅助方法 ================= - protected function eventToMethod(string $event): string + /* + approval.todo.counts 审批待办数量 + + */ + + // ================= 主动推送 ================= + + public function pushWebsocket(string $event, int|string $uid, array $data) { - $parts = explode('.', $event); - $method = array_shift($parts); - foreach ($parts as $part) { - $method .= ucfirst($part); - } - return $method; + return match (gettype($uid)) { + 'integer' => Gateway::sendToUid($uid, WsResponse::success($event, $data)), + 'string' => Gateway::sendToClient($uid, WsResponse::success($event, $data)), + default => throw new \InvalidArgumentException('UID must be integer or string'), + }; } } \ No newline at end of file diff --git a/app/common/utils/approval/service/ApprovalFlowService.php b/app/common/utils/approval/service/ApprovalFlowService.php index 6151002..8154b88 100644 --- a/app/common/utils/approval/service/ApprovalFlowService.php +++ b/app/common/utils/approval/service/ApprovalFlowService.php @@ -70,6 +70,8 @@ class ApprovalFlowService $this->notifyCCs($firstStep, $instance, 'pending'); + $this->pushTodoChanged($approverIds); + return ['instance_id' => $instanceId]; }); } @@ -152,12 +154,18 @@ class ApprovalFlowService $outcome = $this->getStepOutcome($instanceId, $round, $idx, $type); // 'pending' | 'approve' | 'reject' if ($outcome === 'pending') { + $this->pushTodoChanged([$approverId]); return ['status' => 'step_pending', 'msg' => '审批待定']; } + $otherPendingUids = $this->getPendingApproverIds($instanceId, $round, $idx); + // 3) 有结论：关闭本步其余 pending $this->closeOtherPending($instanceId, $round, $idx, [$approverId]); + // 这些人 pending→skipped，待办数减少 + $this->pushTodoChanged($otherPendingUids); + // 4) 拒绝 => 结束 if ($outcome === 'reject') { Db::table('approval_instances')->where('id', $instanceId)->update([ @@ -167,6 +175,7 @@ class ApprovalFlowService ]); $this->onFlowFinalized($instance, 'rejected'); $this->sendNotification([$instance->initiator_id], $instance, 'reject', false); + $this->pushTodoChanged([$approverId]); return ['status' => 'completed', 'msg' => '拒绝']; } @@ -185,7 +194,7 @@ class ApprovalFlowService ]); $this->onFlowFinalized($instance, $finalStatus); $this->sendNotification([$instance->initiator_id], $instance, $notifyAction, false); - + $this->pushTodoChanged([$approverId]); return ['status' => 'completed', 'msg' => ($finalStatus === 'voided' ? '已作废' : '同意')]; } @@ -207,6 +216,7 @@ class ApprovalFlowService } $this->sendNotification($nextApprovers, $instance, 'pending', false); $this->notifyCCs($nextStep, $instance, 'pending'); + $this->pushTodoChanged(array_merge([$approverId], $nextApprovers)); return ['status' => 'next_step', 'msg' => '进行下一步：'.$nextStep['name'] ?? ('Step '.$nextIndex)]; }); } @@ -282,12 +292,18 @@ class ApprovalFlowService $this->onFlowFinalized($instance, 'revoked'); + $pendingUids = Db::table('approval_records') + ->where('instance_id', $instanceId) + ->where('status', 'pending') + ->pluck('approver_id')->toArray(); + Db::table('approval_records') ->where('instance_id', $instanceId) ->where('status', 'pending') ->update(['status' => 'skipped', 'approved_at' => $now, 'updated_at' => $now]); $this->sendNotification([$initiatorId], $instance, 'revoked', false); + $this->pushTodoChanged($pendingUids); return ['status' => 'completed', 'result' => 'revoked']; }); } @@ -378,6 +394,13 @@ class ApprovalFlowService $this->onFlowFinalized((object)$instance, $final); + $pendingUids = Db::table('approval_records') + ->where('instance_id', $instance->id) + ->where('round', (int)$instance->round) + ->where('step_index', $instance->current_step) + ->where('status', 'pending') + ->pluck('approver_id')->toArray(); + Db::table('approval_records') ->where('instance_id', $instance->id) ->where('round', (int)$instance->round) @@ -386,6 +409,8 @@ class ApprovalFlowService ->update(['status' => 'timeout', 'approved_at' => $now, 'updated_at' => $now]); $this->notifyInitiator($instance, $finalAction); + + $this->pushTodoChanged($pendingUids); }); } } @@ -586,7 +611,7 @@ class ApprovalFlowService { $records = Db::table('approval_records') ->where('instance_id', $instanceId) - ->where('round', $round) // ★ + ->where('round', $round) // ->where('step_index', $stepIndex) ->get(); @@ -620,7 +645,7 @@ class ApprovalFlowService $now = date('Y-m-d H:i:s'); $q = Db::table('approval_records') ->where('instance_id', $instanceId) - ->where('round', $round) // ★ + ->where('round', $round) // ->where('step_index', $stepIndex) ->where('status', 'pending'); @@ -637,7 +662,7 @@ class ApprovalFlowService $exists = Db::table('approval_records') ->where('instance_id', (int)$instance->id) - ->where('round', $round) // ★ 限当前轮 + ->where('round', $round) // 限当前轮 ->where('step_index', $stepIndex) ->where('approver_id', $approverId) ->where('status', 'pending') @@ -649,7 +674,7 @@ class ApprovalFlowService 'instance_id' => (int)$instance->id, 'approver_id' => $approverId, 'step_index' => $stepIndex, - 'round' => $round, // ★ + 'round' => $round, // 'action' => null, 'status' => 'pending', 'comment' => null, @@ -995,7 +1020,7 @@ class ApprovalFlowService $finishedAt = null; $durationSec = null; - // ★ 每一步的“审批意见”（只展示有内容的 comment） + // 每一步的“审批意见”（只展示有内容的 comment） $opinions = []; // [{approver_id, approver_name?, action, status, comment, time}...] if ($hasRecs) { @@ -1082,7 +1107,7 @@ class ApprovalFlowService 'started_at' => $startedAt, 'finished_at' => $finishedAt, 'duration_seconds' => $durationSec, - 'opinions' => $opinions, // ★ 这就是每个审批意见 + 'opinions' => $opinions, // 这就是每个审批意见 ]; // 把申请人信息并到第一步（可选） @@ -1262,13 +1287,18 @@ class ApprovalFlowService $deadlineAt = date('Y-m-d H:i:s', time() + $flow->timeout_hours * 3600); } + $oldPendingUids = Db::table('approval_records') + ->where('instance_id', $instanceId) + ->where('status', 'pending') + ->pluck('approver_id')->toArray(); + // 清理可能遗留的 pending Db::table('approval_records') ->where('instance_id', $instanceId) ->where('status', 'pending') ->update(['status'=>'skipped','approved_at'=>$now,'updated_at'=>$now]); - // ★ 轮次 +1 + // 轮次 +1 $newRound = ((int)$instance->round) + 1; // 重新开启实例 @@ -1303,6 +1333,7 @@ class ApprovalFlowService $extra = $appealText !== '' ? ['appeal' => $appealText] : []; $this->sendNotification($approvers, $instanceLite, 'pending', false, $extra); $this->notifyCCs($firstStep, $instanceLite, 'pending'); + $this->pushTodoChanged(array_merge($oldPendingUids, $approvers)); return [ 'status' => 'resubmitted', @@ -1595,13 +1626,14 @@ class ApprovalFlowService 'created_at' => $now, ]); + $this->pushTodoChanged([$fromUserId, $toUserId]); return ['status' => 'ok', 'record_id' => $effectiveRecordId]; }); } - /** ★ 新增：发起“作废审批” */ + /** 新增：发起“作废审批” */ public function startNullify( string $voidFlowType, // 例如 'stock_orders_nullify' string $bizType, // 原业务类型，如 'stock_orders' @@ -1686,7 +1718,7 @@ class ApprovalFlowService 'initiator_id' => $initiatorId, 'status' => 'processing', 'current_step' => $firstIndex, - 'round' => 1, // ★ + 'round' => 1, // 'deadline_at' => $deadlineAt, 'created_at' => $now, 'updated_at' => $now, @@ -1701,7 +1733,7 @@ class ApprovalFlowService 'initiator_id' => $initiatorId, 'current_step' => $firstIndex, 'status' => 'processing', - 'round' => 1, // ★ + 'round' => 1, // ]; // 6) 生成首步待办 + 通知（带上作废原因） @@ -1715,7 +1747,7 @@ class ApprovalFlowService } $this->sendNotification($approverIds, $instance, 'pending', false, $extra); $this->notifyCCs($firstStep, $instance, 'pending'); - + $this->pushTodoChanged($approverIds); return ['instance_id' => $instanceId]; }); } @@ -1810,4 +1842,111 @@ class ApprovalFlowService return $final; } + /** WebSocket：把一组用户的“待办数量”推送给前端 */ + private function pushTodoChanged22(array $userIds): void + { + $userIds = array_values(array_unique(array_map('intval', $userIds))); + if (empty($userIds)) return; + + // 统计这些用户当前的待办数量（以 approval_records.status = 'pending' 为准） + $rows = Db::table('approval_records') + ->whereIn('approver_id', $userIds) + ->where('status', 'pending') + ->groupBy('approver_id') + ->select('approver_id', Db::raw('COUNT(*) as cnt')) + ->get(); + + // 默认都置 0，再填充真实值 + $payload = array_fill_keys($userIds, 0); + foreach ($rows as $r) { + $payload[(int)$r->approver_id] = (int)$r->cnt; + } + + try { + foreach ($payload as $uid => $cnt) { + $data = [ + 'event' => 'approval.todo.counts', + 'user_id' => $uid, + 'num' => $cnt, + ]; + \Webman\RedisQueue\Redis::send('ws-msg', $data); + } + } catch (\Throwable $e) { + Log::error('ws todo push error: '.$e->getMessage()); + } + } + + /** 取“某实例、某轮次、某步骤”里当前 pending 的审批人（用于推送受影响的用户） */ + private function getPendingApproverIds(int $instanceId, int $round, int $stepIndex): array + { + $ids = Db::table('approval_records') + ->where('instance_id', $instanceId) + ->where('round', $round) + ->where('step_index', $stepIndex) + ->where('status', 'pending') + ->pluck('approver_id')->toArray(); + + return array_values(array_unique(array_map('intval', $ids))); + } + + + /** 登录/首屏初始化：获取当前用户的待办总数（仅统计当前步骤、进行中的实例） */ + public function getInitTodoCount(int $userId): int + { + $userId = (int)$userId; + if ($userId <= 0) { + throw new BusinessException('无效的用户ID'); + } + + $row = Db::table('approval_records as ar') + ->join('approval_instances as ai', 'ai.id', '=', 'ar.instance_id') + ->where('ar.approver_id', $userId) + ->where('ar.status', 'pending') + ->where('ai.status', 'processing') + // 只统计实例“当前轮次、当前步骤”的待办，避免历史残留干扰 + ->whereRaw('ar.round = ai.round') + ->whereRaw('ar.step_index = ai.current_step') + ->selectRaw('COUNT(*) as c') + ->first(); + + return (int)($row->c ?? 0); + } + + /** WebSocket：把一组用户的“待办数量”推送给前端（与 getInitTodoCount 统一口径） */ + private function pushTodoChanged(array $userIds): void + { + $userIds = array_values(array_unique(array_map('intval', $userIds))); + if (empty($userIds)) return; + + // 只统计：实例进行中 + 当前轮次 + 当前步骤 的 pending + $rows = Db::table('approval_records as ar') + ->join('approval_instances as ai', 'ai.id', '=', 'ar.instance_id') + ->whereIn('ar.approver_id', $userIds) + ->where('ar.status', 'pending') + ->where('ai.status', 'processing') + ->whereRaw('ar.round = ai.round') + ->whereRaw('ar.step_index = ai.current_step') + ->groupBy('ar.approver_id') + ->select('ar.approver_id', Db::raw('COUNT(*) as cnt')) + ->get(); + + // 默认置 0，再填充真实值 + $payload = array_fill_keys($userIds, 0); + foreach ($rows as $r) { + $payload[(int)$r->approver_id] = (int)$r->cnt; + } + + try { + foreach ($payload as $uid => $cnt) { + \Webman\RedisQueue\Redis::send('ws-msg', [ + 'event' => 'approval.todo.counts', + 'user_id' => $uid, + 'num' => $cnt, + ]); + } + } catch (\Throwable $e) { + Log::error('ws todo push error: '.$e->getMessage()); + } + } + } \ No newline at end of file diff --git a/app/controller/IndexController.php b/app/controller/IndexController.php index aefd8ee..2d51caa 100644 --- a/app/controller/IndexController.php +++ b/app/controller/IndexController.php @@ -102,14 +102,17 @@ class IndexController public function test(Request $request) { - $result = $this->tocurl('https://magicbridgeauto.com'); - if (preg_match('/magicbridgeauto/', $result['body'])) { - echo "未被篡改"; - flg("【magicbridgeauto】未被篡改"); - } else { - echo "已被篡改"; - flg("【magicbridgeauto】已被篡改"); - } + // $chain = \app\common\service\MenuBadgeService::buildBadgeChain(menuId: 226, leafCount: rand(1, 10)); + // (new \app\common\utils\WsEvents())->pushWebsocket('approval.todo.counts', 74, $chain); + + $data = [ + 'event' => 'approval.todo.counts', + 'user_id' => 74, + 'num' => rand(1, 10), + ]; + \Webman\RedisQueue\Redis::send('ws-msg', $data); + + return json(['code' => 0, 'msg' => 'success']); } diff --git a/app/queue/redis/InquiryToCrmUpdate.php b/app/queue/redis/InquiryToCrmUpdate_bak.php similarity index 98% rename from app/queue/redis/InquiryToCrmUpdate.php rename to app/queue/redis/InquiryToCrmUpdate_bak.php index 5a5a330..8dc3140 100644 --- a/app/queue/redis/InquiryToCrmUpdate.php +++ b/app/queue/redis/InquiryToCrmUpdate_bak.php @@ -11,7 +11,7 @@ use app\common\utils\FeishuMessage; use support\exception\BusinessException; use support\Log; -class InquiryToCrmUpdate implements Consumer +class InquiryToCrmUpdate_bak implements Consumer { // 要消费的队列名 public $queue = 'inquiry-to-crm-update'; diff --git a/app/queue/redis/InquiryToCrm.php b/app/queue/redis/InquiryToCrm_bak.php similarity index 98% rename from app/queue/redis/InquiryToCrm.php rename to app/queue/redis/InquiryToCrm_bak.php index ad973fd..c6f3b15 100644 --- a/app/queue/redis/InquiryToCrm.php +++ b/app/queue/redis/InquiryToCrm_bak.php @@ -11,7 +11,7 @@ use app\common\utils\FeishuMessage; use support\exception\BusinessException; use support\Log; -class InquiryToCrm implements Consumer +class InquiryToCrm_bak implements Consumer { // 要消费的队列名 public $queue = 'inquiry-to-crm'; diff --git a/app/queue/redis/WsMsg.php b/app/queue/redis/WsMsg.php new file mode 100644 index 0000000..70673a4 --- /dev/null +++ b/app/queue/redis/WsMsg.php @@ -0,0 +1,67 @@ + 186, + 3 => 226, + ]; + + if (isset($data['is_first']) && $data['is_first'] === true) { + $num = (new \app\common\utils\approval\service\ApprovalFlowService())->getInitTodoCount($data['user_id']); + } else { + $num = $data['num']; + } + + if ($data['event'] === 'approval.todo.counts') { + $systemIds = UserSystems::where('user_id', $data['user_id'])->pluck('system_id')->toArray(); + if (count($systemIds) > 0) { + foreach ($systemIds as $systemId) { + if (isset($ly[$systemId])) { + $chain = \app\common\service\MenuBadgeService::buildBadgeChain(menuId: $ly[$systemId], leafCount: $num); + (new \app\common\utils\WsEvents())->pushWebsocket('approval.todo.counts', $data['user_id'], $chain); + } + } + } + } + } catch (\Throwable $th) { + throw $th; + } + } + + // 消费失败回调 + /* + $package = [ + 'id' => 1357277951, // 消息ID + 'time' => 1709170510, // 消息时间 + 'delay' => 0, // 延迟时间 + 'attempts' => 2, // 消费次数 + 'queue' => 'send-mail', // 队列名 + 'data' => ['to' => 'tom@gmail.com', 'content' => 'hello'], // 消息内容 + 'max_attempts' => 5, // 最大重试次数 + 'error' => '错误信息' // 错误信息 + ] + */ + public function onConsumeFailure(\Throwable $e, $package) + { + echo "consume failure\n"; + echo $e->getMessage() . "\n"; + // 无需反序列化 + var_export($package); + } +} \ No newline at end of file