setName('card:queue') ->setDescription('Process messages from the Redis queue'); } protected function execute(Input $input, Output $output) { $this->queueService = new QueueService(); while (true) { $message = $this->queueService->pop('card_queue'); if ($message) { // 处理消息 $this->processMessage($message); } else { // 暂停一段时间再检查队列 sleep(1); } } } protected function processMessage($message) { $SybOil = new SybOil(); // 拿到队列数据处理 $data = json_decode($message, true); $SybOil->saveLogStore('card_queue队列数据处理:【'.$message.'】'); $userId = $data['user_id']; $cardId = $data['card_id']; // 查询兑换卡 $card = Db::name('card') ->where('card_id', $cardId) ->find(); $field = [ 'card_id'=>$cardId, 'card_number'=>$card['card_number'], 'user_id'=>$userId, 'money'=>$card['amount'], 'store_id'=>$card['store_id'], ]; if ($card['is_used'] != 10) { CardLogModel::add(array_merge( [ 'state'=>10, 'describe'=>'实体兑换卡已使用', ], $field )); return false; // 无效的卡或密钥 } Db::startTrans(); try { // 更新兑换卡使用状态 $affectedRows = Db::name('card') ->where('card_id', $card['card_id']) ->where('version', $card['version']) ->update([ 'is_used' => 20, 'used_time' => time(), 'user_id' => $userId, 'version' => $card['version'] + 1, ]); if ($affectedRows === 0) { Db::rollback(); // 增加兑换记录 CardLogModel::add(array_merge( [ 'state'=>10, 'describe'=>'兑换失败,已使用', ], $field )); return false; } // 更新用户余额 $obj = UserModel::detail($userId); // 增加分销对象的余额 $inc = $obj->exchangeToBalance($obj['mobile'],[ 'money'=>$card['amount'], 'mode'=>'inc', 'remark'=>'实体卡兑换', ], $card['store_id']); if(!$inc){ Db::rollback(); CardLogModel::add(array_merge( [ 'state'=>10, 'describe'=>'增加用户余额失败', ], $field )); return false; } // 增加兑换记录 CardLogModel::add(array_merge( [ 'state'=>20, 'describe'=>'兑换成功', ], $field )); Db::commit(); return true; } catch (\Exception $e) { $SybOil->saveLogStore($e->getMessage()); Db::rollback(); return false; } } }