im/app/queue/single/Power.php

<?php

namespace app\queue\single;

use app\model\BalanceLog;
use Webman\RedisQueue\Consumer;
use app\model\WorkRecord;
use think\facade\Db;

class Power implements Consumer
{
    // 要消费的队列名
    public $queue = 'Power';
    public $connection = 'default';

    // 消费Notify
    public function consume($data)
    {
        return false;
        try {
            if($data['action'] == 'power_expris'){
                $this->log('开始失效算力:'.json_encode($data));
                $log = (new \app\model\BalanceLog)->setSuffix('_currency1')
                    ->where('id',$data['id'])
                    ->where('status',1)
                    ->find();
                if(!$log){
                    return ;
                }
                Db::startTrans();
                try {
                    $log->status = 2;
                    $log->save();
                    \app\model\User::transform(
                        'currency1',
                        'currency2',
                        $log->user_id,
                        $log->amount,
                        \app\enum\BalanceType::POWER_EXPRIS,
                        $log->_id?:$log->id
                    );
                    Db::commit();
                } catch (\Exception $e) {
                    Db::rollback();
                    $this->log($e->getMessage());
                    throw $e;
                }
            }elseif($data['action'] == 'realese'){
                $this->log('开始释放算力:'.json_encode($data));
                $user = \app\model\User::find($data['user_id']);
                $list = (new BalanceLog)->setSuffix('_currency1')
                    ->where('user_id',intval($data['user_id']))
                    ->where('status','<>',2)
                    ->where('amount','>',0)
                    ->order('created_at','asc')
                    ->select();
                $this->log($list->toArray());
                //需要结算的记录
                $_calcs = [];
                //用户的待结算余额
                $_currency1 = $user->currency1;
                //用户的算力余额
                $_score = $user->score;
                foreach ($list as $key => $vo) { 
                    //如果用户的待结算余额和算力都大于本记录的金额,就结算本条否则就结束计算
                    if($_currency1 - $vo->amount >= 0 && $_score-$vo->amount>=0){
                        //更新剩余货币
                        $_currency1-=$vo->amount;
                        $_score-=$vo->amount;
                        $_calcs[]=$vo;
                        continue;
                    }else{
                        break;
                    }
                }
                Db::startTrans();
                try {
                    $transfrom_amount = 0;
                    foreach($_calcs as $k=>$log){
                        $transfrom_amount += $log->amount;
                        $log->status = 2;
                        $log->save();
                    }
                    if($transfrom_amount != 0){
                        \app\model\User::transform('score','money',$user->id,$transfrom_amount,\app\enum\BalanceType::POWER_REALESE);
                        \app\model\User::currency1($user->id,-$transfrom_amount,\app\enum\BalanceType::POWER_REALESE);
                        cache_add('user_income_total_'.$user->user_id,$transfrom_amount);
                    }
                    Db::commit();
                } catch (\Exception $e) {
                    $this->log($e->getMessage());
                    Db::rollback();
                }
            }
        }
        catch(\Exception $e)
        {
            $this->log($e->getMessage());
        }
    }
    function log($msg = ''){
        log_alert($msg,'power');
    }
    // 消费失败回调
    /* 
    $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)
    {
        $this->log('consume failure:'.$e->getMessage());
    }
}