redis-queue多进程下消费如何保证不重复入库的问题

1. 数据加悲观锁
2. 加唯一索引，处理异常（更新）
3. 提前建好，队列直接更新操作

$transferData = self::$transferData;
        $project = $transferData['project'];
        $time = date('Y-m-d H:i:s');
        try{
            Db::table('project')->insert([
                'project'    => $transferData['project'],
                'created_at' => $time,
                'updated_at' => $time
            ]);
        }catch (\Throwable $th){
            Db::table('project')->where('project', $project)->update([
                'updated_at' => $time
            ]);
        }

表中加入了唯一索引，代码调成这种后会出现mysql死锁的情况

安装插件：https://www.workerman.net/plugin/55

创建统计锁

/**
 * 统计锁
 * @method static LockInterface lock(string $day, ?float $ttl = null, ?bool $autoRelease = null) 数据库操作锁
 */
class StatisticLocker extends Locker
{
}

使用锁

//取锁：有效期3秒，自释放
$lock = \app\services\lock\StatisticLocker::lock($day, 3, true);
//多进程阻塞等待
if ($lock->acquire(true)) {
    $row = Db::table('statistic')->where('day', $day)->first();
    if ($row) {
        $data = [
            'count'         => $row->count + 1,
            'cost'          => $row->cost + $costTime,
            'success_count' => $row->success_count + ($success ? 1 : 0),
            'error_count'   => $row->error_count + ($success ? 0 : 1),
        ];
        Db::table('statistic')->where('day', $day)->update($data);
    } else {
        $data = [
            'day'           => $day,
            'count'         => 1,
            'cost'          => $costTime,
            'success_count' => $success ? 1 : 0,
            'error_count'   => $success ? 0 : 1,
        ];
        Db::table('statistic')->insert($data);
    }
}
//代码执行到这里，会立刻自动释放锁

class SyncStatisticsToMysql implements Consumer
{
    // 要消费的队列名
    public string $queue = 'sync-statistics-to-mysql';

    // 连接名，对应 plugin/webman/redis-queue/redis.php 里的连接`
    public string $connection = 'default';

    // 消费的数据
    public static array $transferData = [];

    // 消费
    public function consume($data)
    {
        self::$transferData = $data;
        Db::beginTransaction();
        try {
            // 组装数据
            self::assembleData();
            // 应用
            self::project();
            // 调用记录
            self::tracing();
            // 总统计
            self::statistic();
            // 应用总统计
            self::statisticProject();
            // 应用每分钟统计
            self::statisticProjectInterval();
            // 应用调用统计
            self::statisticProjectTransfer();
            // 应用调用每分钟统计
            self::statisticProjectTransferInterval();
            // 应用状态码统计
            self::statisticProjectCode();
            // 应用状态码每分钟统计
            self::statisticProjectCodeInterval();
            // 应用IP统计
            self::statisticProjectIp();
            // 应用IP每分钟统计
            self::statisticProjectIpInterval();
            // 应用IP状态码统计
            self::statisticProjectIpCode();
            // 应用IP状态码每分钟统计
            self::statisticProjectIpCodeInterval();
            // 应用IP调用统计
            self::statisticProjectIpTransfer();
            // 应用IP调用每分钟统计
            self::statisticProjectIpTransferInterval();
            Db::commit();
        } catch (\Throwable $th) {
            Db::rollBack();
            var_dump($th->getMessage());
            throw new ServerErrorHttpException($th->getMessage());
        }
    }

    /**
     * 组装数据
     */
    protected static function assembleData()
    {
        $transferData = self::$transferData;
        $names = explode('-', $transferData['project']);
        $transferData['project'] = $names[0];
        $transferData['type'] = $names[1] ?? 'All';
        $timer = strtotime($transferData['time']);
        $transferData['day'] = date('Ymd', $timer);
        $transferData['minute'] = date('YmdHi', $timer);
        $transferData['trace'] = uniqid();
        self::$transferData = $transferData;
    }

    /**
     * 应用
     */
    protected static function project()
    {
        $transferData = self::$transferData;
        $project = $transferData['project'];
        $time = date('Y-m-d H:i:s');
        try {
            Db::table('project')->insert([
                'project'    => $transferData['project'],
                'created_at' => $time,
                'updated_at' => $time
            ]);
        } catch (\PDOException | \Exception $e) {
            Db::table('project')->where('project', $project)->update([
                'updated_at' => $time
            ]);
        }
    }

    /**
     * 调用记录
     */
    protected static function tracing()
    {
        $transferData = self::$transferData;
        $insert = [
            'day'       => $transferData['day'],
            'time'      => $transferData['time'],
            'trace'     => $transferData['trace'],
            'project'   => $transferData['project'],
            'type'      => $transferData['type'],
            'ip'        => $transferData['ip'],
            'transfer'  => $transferData['transfer'],
            'cost_time' => $transferData['costTime'],
            'success'   => $transferData['success'],
            'code'      => $transferData['code'],
            'details'   => $transferData['details'],
        ];
        Db::table('tracing')->insert($insert);
    }

    /**
     * 总统计
     */
    public static function statistic()
    {
        $transferData = self::$transferData;
        $day = $transferData['day'];
        $success = 1 === $transferData['success'];
        $costTime = $transferData['costTime'];
        try {
            $data = [
                'day'           => $day,
                'count'         => 1,
                'cost'          => $costTime,
                'success_count' => $success ? 1 : 0,
                'error_count'   => $success ? 0 : 1,
            ];
            Db::table('statistic')->insert($data);
        } catch (\PDOException | \Exception $e) {
            $data = [
                'count'         => Db::raw('count + 1'),
                'cost'          => Db::raw("cost + {$costTime}"),
            ];
            if($success){
                $data['success_count'] = Db::raw('success_count + 1');
            }else{
                $data['error_count'] = Db::raw('error_count + 1');
            }
            Db::table('statistic')->where('day', $day)
                ->update($data);
        }
    }

    /**
     * 应用总统计
     */
    public static function statisticProject()
    {
        $transferData = self::$transferData;
        $day = $transferData['day'];
        $project = $transferData['project'];
        $type = $transferData['type'];
        $success = 1 === $transferData['success'];
        $costTime = $transferData['costTime'];
        try {
            $data = [
                'day'           => $day,
                'project'       => $project,
                'type'          => $type,
                'count'         => 1,
                'cost'          => $costTime,
                'success_count' => $success ? 1 : 0,
                'error_count'   => $success ? 0 : 1,
            ];
            Db::table('statistic_project')->insert($data);
        } catch (\PDOException | \Exception $e) {
            $where = [
                ['day', '=', $day],
                ['project', '=', $project],
                ['type', '=', $type],
            ];
            $data = [
                'count'         => Db::raw('count + 1'),
                'cost'          => Db::raw("cost + {$costTime}"),
            ];
            if($success){
                $data['success_count'] = Db::raw('success_count + 1');
            }else{
                $data['error_count'] = Db::raw('error_count + 1');
            }
            Db::table('statistic_project')
                ->where($where)
                ->update($data);
        }
    }

    /**
     * 应用每分钟统计
     */
    public static function statisticProjectInterval()
    {
        $transferData = self::$transferData;
        $day = $transferData['day'];
        $minute = $transferData['minute'];
        $project = $transferData['project'];
        $type = $transferData['type'];
        $success = 1 === $transferData['success'];
        $costTime = $transferData['costTime'];
        try {
            $data = [
                'day'           => $day,
                'time'          => $minute,
                'project'       => $project,
                'type'          => $type,
                'count'         => 1,
                'cost'          => $costTime,
                'success_count' => $success ? 1 : 0,
                'error_count'   => $success ? 0 : 1,
            ];
            Db::table('statistic_project_interval')->insert($data);
        } catch (\PDOException | \Exception $e) {
            $where = [
                ['day', '=', $day],
                ['time', '=', $minute],
                ['project', '=', $project],
                ['type', '=', $type],
            ];
            $data = [
                'count'         => Db::raw('count + 1'),
                'cost'          => Db::raw("cost + {$costTime}"),
            ];
            if($success){
                $data['success_count'] = Db::raw('success_count + 1');
            }else{
                $data['error_count'] = Db::raw('error_count + 1');
            }
            Db::table('statistic_project_interval')
                ->where($where)
                ->update($data);
        }
    }

    /**
     * 应用调用统计
     */
    public static function statisticProjectTransfer()
    {
        $transferData = self::$transferData;
        $day = $transferData['day'];
        $project = $transferData['project'];
        $type = $transferData['type'];
        $costTime = $transferData['costTime'];
        $transfer = $transferData['transfer'];
        $success = 1 === $transferData['success'];
        try {
            $data = [
                'day'           => $day,
                'project'       => $project,
                'type'          => $type,
                'transfer'      => $transfer,
                'count'         => 1,
                'cost'          => $costTime,
                'success_count' => $success ? 1 : 0,
                'error_count'   => $success ? 0 : 1,
            ];
            Db::table('statistic_project_transfer')->insert($data);
        } catch (\PDOException | \Exception $e) {
            $where = [
                ['day', '=', $day],
                ['project', '=', $project],
                ['type', '=', $type],
                ['transfer', '=', $transfer]
            ];
            $data = [
                'count'         => Db::raw('count + 1'),
                'cost'          => Db::raw("cost + {$costTime}"),
            ];
            if($success){
                $data['success_count'] = Db::raw('success_count + 1');
            }else{
                $data['error_count'] = Db::raw('error_count + 1');
            }
            Db::table('statistic_project_transfer')
                ->where($where)
                ->update($data);
        }
    }

    /**
     * 应用调用每分钟统计
     */
    public static function statisticProjectTransferInterval()
    {
        $transferData = self::$transferData;
        $day = $transferData['day'];
        $minute = $transferData['minute'];
        $project = $transferData['project'];
        $type = $transferData['type'];
        $costTime = $transferData['costTime'];
        $transfer = $transferData['transfer'];
        $success = 1 === $transferData['success'];
        try {
            $data = [
                'day'           => $day,
                'time'          => $minute,
                'project'       => $project,
                'type'          => $type,
                'transfer'      => $transfer,
                'count'         => 1,
                'cost'          => $costTime,
                'success_count' => $success ? 1 : 0,
                'error_count'   => $success ? 0 : 1,
            ];
            Db::table('statistic_project_transfer_interval')->insert($data);
        } catch (\PDOException | \Exception $e) {
            $where = [
                ['day', '=', $day],
                ['time', '=', $minute],
                ['project', '=', $project],
                ['type', '=', $type],
                ['transfer', '=', $transfer]
            ];
            $data = [
                'count'         => Db::raw('count + 1'),
                'cost'          => Db::raw("cost + {$costTime}"),
            ];
            if($success){
                $data['success_count'] = Db::raw('success_count + 1');
            }else{
                $data['error_count'] = Db::raw('error_count + 1');
            }
            Db::table('statistic_project_ip_transfer_interval')
                ->where($where)
                ->update($data);
        }
    }

    /**
     * 应用状态码统计
     */
    public static function statisticProjectCode()
    {
        $transferData = self::$transferData;
        $day = $transferData['day'];
        $project = $transferData['project'];
        $type = $transferData['type'];
        $costTime = $transferData['costTime'];
        $code = $transferData['code'];
        try {
            $data = [
                'day'     => $day,
                'project' => $project,
                'type'    => $type,
                'code'    => $code,
                'count'   => 1,
                'cost'    => $costTime,
            ];
            Db::table('statistic_project_code')->insert($data);
        } catch (\PDOException | \Exception $e) {
            $where = [
                ['day', '=', $day],
                ['code', '=', $code],
                ['project', '=', $project],
                ['type', '=', $type]
            ];
            Db::table('statistic_project_code')
                ->where($where)
                ->update([
                    'count' => Db::raw('count + 1'),
                    'cost'  => Db::raw("cost + {$costTime}"),
                ]);
        }
    }

    /**
     * 应用状态码每分钟统计
     */
    public static function statisticProjectCodeInterval()
    {
        $transferData = self::$transferData;
        $day = $transferData['day'];
        $minute = $transferData['minute'];
        $project = $transferData['project'];
        $type = $transferData['type'];
        $costTime = $transferData['costTime'];
        $code = $transferData['code'];
        try {
            $data = [
                'day'     => $day,
                'time'    => $minute,
                'project' => $project,
                'type'    => $type,
                'code'    => $code,
                'count'   => 1,
                'cost'    => $costTime,
            ];
            Db::table('statistic_project_code_interval')->insert($data);
        } catch (\PDOException | \Exception $e) {
            $where = [
                ['day', '=', $day],
                ['time', '=', $minute],
                ['code', '=', $code],
                ['project', '=', $project],
                ['type', '=', $type]
            ];
            Db::table('statistic_project_code_interval')
                ->where($where)
                ->update([
                    'count' => Db::raw('count + 1'),
                    'cost'  => Db::raw("cost + {$costTime}"),
                ]);
        }
    }

    /**
     * 应用IP统计
     */
    public static function statisticProjectIp()
    {
        $transferData = self::$transferData;
        $day = $transferData['day'];
        $project = $transferData['project'];
        $type = $transferData['type'];
        $costTime = $transferData['costTime'];
        $success = 1 === $transferData['success'];
        $ip = $transferData['ip'];
        try {
            $data = [
                'day'           => $day,
                'project'       => $project,
                'type'          => $type,
                'ip'            => $ip,
                'count'         => 1,
                'cost'          => $costTime,
                'success_count' => $success ? 1 : 0,
                'error_count'   => $success ? 0 : 1,
            ];
            Db::table('statistic_project_ip')->insert($data);
        } catch (\PDOException | \Exception $e) {
            $where = [
                ['day', '=', $day],
                ['project', '=', $project],
                ['type', '=', $type],
                ['ip', '=', $ip],
            ];

            $data = [
                'count'         => Db::raw('count + 1'),
                'cost'          => Db::raw("cost + {$costTime}"),
            ];
            if($success){
                $data['success_count'] = Db::raw('success_count + 1');
            }else{
                $data['error_count'] = Db::raw('error_count + 1');
            }
            Db::table('statistic_project_ip')
                ->where($where)
                ->update($data);
        }
    }

    /**
     * 应用IP每分钟统计
     */
    public static function statisticProjectIpInterval()
    {
        $transferData = self::$transferData;
        $day = $transferData['day'];
        $minute = $transferData['minute'];
        $project = $transferData['project'];
        $type = $transferData['type'];
        $costTime = $transferData['costTime'];
        $success = 1 === $transferData['success'];
        $ip = $transferData['ip'];

        try {
            $data = [
                'day'           => $day,
                'time'          => $minute,
                'project'       => $project,
                'type'          => $type,
                'ip'            => $ip,
                'count'         => 1,
                'cost'          => $costTime,
                'success_count' => $success ? 1 : 0,
                'error_count'   => $success ? 0 : 1,
            ];
            Db::table('statistic_project_ip_interval')->insert($data);
        } catch (\PDOException | \Exception $e) {
            $where = [
                ['day', '=', $day],
                ['time', '=', $minute],
                ['project', '=', $project],
                ['type', '=', $type],
                ['ip', '=', $ip],
            ];

            $data = [
                'count'         => Db::raw('count + 1'),
                'cost'          => Db::raw("cost + {$costTime}"),
            ];
            if($success){
                $data['success_count'] = Db::raw('success_count + 1');
            }else{
                $data['error_count'] = Db::raw('error_count + 1');
            }
            Db::table('statistic_project_ip_interval')
                ->where($where)
                ->update($data);
        }
    }

    /**
     * 应用IP状态码统计
     */
    public static function statisticProjectIpCode()
    {
        $transferData = self::$transferData;
        $day = $transferData['day'];
        $project = $transferData['project'];
        $type = $transferData['type'];
        $costTime = $transferData['costTime'];
        $ip = $transferData['ip'];
        $code = $transferData['code'];
        try {
            $data = [
                'day'     => $day,
                'project' => $project,
                'type'    => $type,
                'ip'      => $ip,
                'code'    => $code,
                'count'   => 1,
                'cost'    => $costTime,
            ];
            Db::table('statistic_project_ip_code')->insert($data);
        } catch (\PDOException | \Exception $e) {
            $where = [
                ['day', '=', $day],
                ['code', '=', $code],
                ['project', '=', $project],
                ['type', '=', $type],
                ['ip', '=', $ip]
            ];
            Db::table('statistic_project_ip_code')
                ->where($where)
                ->update([
                    'count' => Db::raw('count + 1'),
                    'cost'  => Db::raw("cost + {$costTime}"),
                ]);
        }
    }

    /**
     * 应用IP状态码每分钟统计
     */
    public static function statisticProjectIpCodeInterval()
    {
        $transferData = self::$transferData;
        $day = $transferData['day'];
        $minute = $transferData['minute'];
        $project = $transferData['project'];
        $type = $transferData['type'];
        $costTime = $transferData['costTime'];
        $ip = $transferData['ip'];
        $code = $transferData['code'];

        try {
            $data = [
                'day'     => $day,
                'time'    => $minute,
                'project' => $project,
                'type'    => $type,
                'ip'      => $ip,
                'code'    => $code,
                'count'   => 1,
                'cost'    => $costTime,
            ];
            Db::table('statistic_project_ip_code_interval')->insert($data);
        } catch (\PDOException | \Exception $e) {
            $where = [
                ['day', '=', $day],
                ['time', '=', $minute],
                ['project', '=', $project],
                ['type', '=', $type],
                ['ip', '=', $ip],
                ['code', '=', $code],
            ];
            Db::table('statistic_project_ip_code_interval')
                ->where($where)
                ->update([
                    'count' => Db::raw('count + 1'),
                    'cost'  => Db::raw("cost + {$costTime}"),
                ]);
        }
    }

    /**
     * 应用IP调用统计
     */
    public static function statisticProjectIpTransfer()
    {
        $transferData = self::$transferData;
        $day = $transferData['day'];
        $project = $transferData['project'];
        $type = $transferData['type'];
        $costTime = $transferData['costTime'];
        $ip = $transferData['ip'];
        $transfer = $transferData['transfer'];
        $success = 1 === $transferData['success'];

        try {
            $data = [
                'day'           => $day,
                'project'       => $project,
                'type'          => $type,
                'ip'            => $ip,
                'transfer'      => $transfer,
                'count'         => 1,
                'cost'          => $costTime,
                'success_count' => $success ? 1 : 0,
                'error_count'   => $success ? 0 : 1,
            ];
            Db::table('statistic_project_ip_transfer')->insert($data);
        } catch (\PDOException | \Exception $e) {
            $where = [
                ['day', '=', $day],
                ['project', '=', $project],
                ['type', '=', $type],
                ['ip', '=', $ip],
                ['transfer', '=', $transfer]
            ];
            $data = [
                'count'         => Db::raw('count + 1'),
                'cost'          => Db::raw("cost + {$costTime}"),
            ];
            if($success){
                $data['success_count'] = Db::raw('success_count + 1');
            }else{
                $data['error_count'] = Db::raw('error_count + 1');
            }
            Db::table('statistic_project_ip_transfer')
                ->where($where)
                ->update($data);
        }
    }

    /**
     * 应用IP调用每分钟统计
     */
    public static function statisticProjectIpTransferInterval()
    {
        $transferData = self::$transferData;
        $day = $transferData['day'];
        $minute = $transferData['minute'];
        $project = $transferData['project'];
        $type = $transferData['type'];
        $costTime = $transferData['costTime'];
        $ip = $transferData['ip'];
        $transfer = $transferData['transfer'];
        $success = 1 === $transferData['success'];

        try {
            $data = [
                'day'           => $day,
                'time'          => $minute,
                'project'       => $project,
                'type'          => $type,
                'ip'            => $ip,
                'transfer'      => $transfer,
                'count'         => 1,
                'cost'          => $costTime,
                'success_count' => $success ? 1 : 0,
                'error_count'   => $success ? 0 : 1,
            ];
            Db::table('statistic_project_ip_transfer_interval')->insert($data);
        } catch (\PDOException | \Exception $e) {
            $where = [
                ['day', '=', $day],
                ['time', '=', $minute],
                ['project', '=', $project],
                ['type', '=', $type],
                ['ip', '=', $ip],
                ['transfer', '=', $transfer]
            ];

            $data = [
                'count'         => Db::raw('count + 1'),
                'cost'          => Db::raw("cost + {$costTime}"),
            ];
            if($success){
                $data['success_count'] = Db::raw('success_count + 1');
            }else{
                $data['error_count'] = Db::raw('error_count + 1');
            }
            Db::table('statistic_project_ip_transfer_interval')
                ->where($where)
                ->update($data);
        }
    }
}

有个statistics的统计系统，我抄的这个，但是他的貌似是定时器加redis实现的，我想用队列的多进程

/**
 * 统计锁
 * @method static LockInterface lock(string $day, ?float $ttl = null, ?bool $autoRelease = null) 数据库操作锁
 * @method static LockInterface project(string $project, ?float $ttl = null, ?bool $autoRelease = null) 应用锁
 * @method static LockInterface statistic(string $day, ?float $ttl = null, ?bool $autoRelease = null) 总统计锁
 * @method static LockInterface statisticProject(string $day_project_type, ?float $ttl = null, ?bool $autoRelease = null) 应用总统计
 * @method static LockInterface statisticProjectInterval(string $day_minute_project_type, ?float $ttl = null, ?bool $autoRelease = null) 应用每分钟统计
 * @method static LockInterface statisticProjectTransfer(string $key, ?float $ttl = null, ?bool $autoRelease = null) 应用调用统计
 * @method static LockInterface statisticProjectTransferInterval(string $key, ?float $ttl = null, ?bool $autoRelease = null) 应用调用每分钟统计
 * @method static LockInterface statisticProjectCode(string $key, ?float $ttl = null, ?bool $autoRelease = null) 应用状态码统计
 * @method static LockInterface statisticProjectCodeInterval(string $key, ?float $ttl = null, ?bool $autoRelease = null) 应用状态码每分钟统计
 * @method static LockInterface statisticProjectIp(string $key, ?float $ttl = null, ?bool $autoRelease = null) 应用IP统计
 * @method static LockInterface statisticProjectIpInterval(string $key, ?float $ttl = null, ?bool $autoRelease = null) 应用IP每分钟统计
 * @method static LockInterface statisticProjectIpCode(string $key, ?float $ttl = null, ?bool $autoRelease = null) 应用IP状态码统计
 * @method static LockInterface statisticProjectIpCodeInterval(string $key, ?float $ttl = null, ?bool $autoRelease = null) 应用IP状态码每分钟统计
 * @method static LockInterface statisticProjectIpTransfer(string $key, ?float $ttl = null, ?bool $autoRelease = null) 应用IP调用统计
 * @method static LockInterface statisticProjectIpTransferInterval(string $key, ?float $ttl = null, ?bool $autoRelease = null) 应用IP调用每分钟统计
 */
class StatisticLocker extends Locker
{

}

// 提高性能
if (\support\Redis::exists($key)) {
    goto update_db;
}
//取锁：有效期3秒，自释放
$lock = \app\services\lock\StatisticLocker::lock($day, 3, true);
//多进程阻塞等待
if ($lock->acquire(true)) {
    $row = Db::table('statistic')->where('day', $day)->first();
    if ($row) {
        update_db:
        //这里不要用查询出的递增字段的值，因为多进程下存在并发，数据会不准！！
        //最佳实践，根据查询出的数据，拼接$where，而$data使用数据库原生语法
        Db::table('statistic')->where($where)->update($data);
    } else {
        $data = [
            'day'           => $day,
            'count'         => 1,
            'cost'          => $costTime,
            'success_count' => $success ? 1 : 0,
            'error_count'   => $success ? 0 : 1,
        ];
        Db::table('statistic')->insert($data);
        //设置标志
        \support\Redis::setEx($key, 300, time());
    }
}
//代码执行到这里，会立刻自动释放锁

<?php

namespace app\queue\redis;

use app\services\lock\StatisticLocker;
use support\Db;
use support\Redis;
use Tinywan\ExceptionHandler\Exception\ServerErrorHttpException;
use Webman\RedisQueue\Consumer;

class SyncStatisticsToMysql implements Consumer
{
    // 要消费的队列名
    public string $queue = 'sync-statistics-to-mysql';

    // 连接名，对应 plugin/webman/redis-queue/redis.php 里的连接`
    public string $connection = 'default';

    // 消费的数据
    public static array $transferData = [];

    public static $num = 0;

    // 消费
    public function consume($data)
    {
        $startTime = microtime(true);
        self::$transferData = $data;
        try {
            // 组装数据
            self::assembleData();
            // 调用记录
            self::tracing();
            // 应用
            self::project();
            // 总统计
            self::statistic();
            // 应用总统计
            self::statisticProject();
            // 应用每分钟统计
            self::statisticProjectInterval();
            // 应用调用统计
            self::statisticProjectTransfer();
            // 应用调用每分钟统计
            self::statisticProjectTransferInterval();
            // 应用状态码统计
            self::statisticProjectCode();
            // 应用状态码每分钟统计
            self::statisticProjectCodeInterval();
            // 应用IP统计
            self::statisticProjectIp();
            // 应用IP每分钟统计
            self::statisticProjectIpInterval();
            // 应用IP状态码统计
            self::statisticProjectIpCode();
            // 应用IP状态码每分钟统计
            self::statisticProjectIpCodeInterval();
            // 应用IP调用统计
            self::statisticProjectIpTransfer();
            //应用IP调用每分钟统计
            self::statisticProjectIpTransferInterval();
        } catch (\Throwable $th) {
            var_dump($th->getMessage());
            throw new ServerErrorHttpException($th->getMessage());
        }
        $endTime = microtime(true);
        self::$num += $endTime - $startTime;
        var_dump(self::$num);
    }

    /**
     * 组装数据
     */
    protected static function assembleData()
    {
        $transferData = self::$transferData;
        $names = explode('-', $transferData['project']);
        $transferData['project'] = $names[0];
        $transferData['type'] = $names[1] ?? 'All';
        $timer = strtotime($transferData['time']);
        $transferData['day'] = date('Ymd', $timer);
        $transferData['minute'] = date('YmdHi', $timer);
        $transferData['trace'] = uniqid();
        $transferData['project_key'] = md5($transferData['project']);
        $transferData['statistic_key'] = md5($transferData['day']);
        $transferData['statistic_project_key'] = md5($transferData['day'] . $transferData['project'] . $transferData['type']);
        $transferData['statistic_project_interval_key'] = md5($transferData['day'] . $transferData['project'] . $transferData['type'] . $transferData['minute']);
        $transferData['statistic_project_transfer_key'] = md5($transferData['day'] . $transferData['project'] . $transferData['type'] . $transferData['transfer']);
        $transferData['statistic_project_transfer_interval_key'] = md5($transferData['day'] . $transferData['project'] . $transferData['type'] . $transferData['transfer'] . $transferData['minute']);
        $transferData['statistic_project_code_key'] = md5($transferData['day'] . $transferData['project'] . $transferData['type'] . $transferData['code']);
        $transferData['statistic_project_code_interval_key'] = md5($transferData['day'] . $transferData['project'] . $transferData['type'] . $transferData['code'] . $transferData['minute']);
        $transferData['statistic_project_ip_key'] = md5($transferData['day'] . $transferData['project'] . $transferData['type'] . $transferData['ip']);
        $transferData['statistic_project_ip_interval_key'] = md5($transferData['day'] . $transferData['project'] . $transferData['type'] . $transferData['ip'] . $transferData['minute']);
        $transferData['statistic_project_ip_code_key'] = md5($transferData['day'] . $transferData['project'] . $transferData['type'] . $transferData['ip'] . $transferData['code']);
        $transferData['statistic_project_ip_code_interval_key'] = md5($transferData['day'] . $transferData['project'] . $transferData['type'] . $transferData['ip'] . $transferData['code'] . $transferData['minute']);
        $transferData['statistic_project_ip_transfer_key'] = md5($transferData['day'] . $transferData['project'] . $transferData['type'] . $transferData['ip'] . $transferData['transfer']);
        $transferData['statistic_project_ip_transfer_interval_key'] = md5($transferData['day'] . $transferData['project'] . $transferData['type'] . $transferData['ip'] . $transferData['transfer'] . $transferData['minute']);
        self::$transferData = $transferData;
    }

    /**
     * 调用记录
     */
    protected static function tracing()
    {
        $transferData = self::$transferData;
        $insert = [
            'day'       => $transferData['day'],
            'time'      => $transferData['time'],
            'trace'     => $transferData['trace'],
            'project'   => $transferData['project'],
            'type'      => $transferData['type'],
            'ip'        => $transferData['ip'],
            'transfer'  => $transferData['transfer'],
            'cost_time' => $transferData['costTime'],
            'success'   => $transferData['success'],
            'code'      => $transferData['code'],
            'details'   => $transferData['details'],
        ];
        Db::table('tracing')->insert($insert);
    }

    /**
     * 应用
     */
    protected static function project()
    {
        $transferData = self::$transferData;
        $project = $transferData['project'];
        $time = date('Y-m-d H:i:s');
        $key = $transferData['project_key'];
        if (Redis::exists('key:' . $key)) {
            goto update_db;
        }
        //取锁：有效期3秒，自释放
        $lock = StatisticLocker::lock($key, 3, true);

        //多进程阻塞等待
        if ($lock->acquire(true)) {
            $row = Db::table('project')->where('key', $key)->first();
            if ($row) {
                update_db:
                //这里不要用查询出的递增字段的值，因为多进程下存在并发，数据会不准！！
                //最佳实践，根据查询出的数据，拼接$where，而$data使用数据库原生语法
                Db::table('project')->where('key', $key)->update([
                    'updated_at' => $time
                ]);
            } else {
                $data = [
                    'key'        => $key,
                    'project'    => $project,
                    'created_at' => $time,
                    'updated_at' => $time
                ];
                Db::table('project')->insert($data);
                //设置标志
                Redis::setEx('key:' . $key, 300, time());
            }
        }
        //代码执行到这里，会立刻自动释放锁
    }

    /**
     * 总统计
     */
    public static function statistic()
    {
        $transferData = self::$transferData;
        $day = $transferData['day'];
        $success = 1 === $transferData['success'];
        $costTime = $transferData['costTime'];
        $key = $transferData['statistic_key'];
        if (Redis::exists('key:' . $key)) {
            goto update_db;
        }
        //取锁：有效期3秒，自释放
        $lock = StatisticLocker::lock($key, 3, true);

        //多进程阻塞等待
        if ($lock->acquire(true)) {
            $row = Db::table('statistic')->where('key', $key)->first();
            if ($row) {
                update_db:
                //这里不要用查询出的递增字段的值，因为多进程下存在并发，数据会不准！！
                //最佳实践，根据查询出的数据，拼接$where，而$data使用数据库原生语法
                $data = [
                    'count' => Db::raw('count + 1'),
                    'cost'  => Db::raw("cost + {$costTime}"),
                ];
                if ($success) {
                    $data['success_count'] = Db::raw('success_count + 1');
                } else {
                    $data['error_count'] = Db::raw('error_count + 1');
                }
                Db::table('statistic')->where('key', $key)
                    ->update($data);
            } else {
                $data = [
                    'key'           => $key,
                    'day'           => $day,
                    'count'         => 1,
                    'cost'          => $costTime,
                    'success_count' => $success ? 1 : 0,
                    'error_count'   => $success ? 0 : 1,
                ];
                Db::table('statistic')->insert($data);
                //设置标志
                Redis::setEx('key:' . $key, 300, time());
            }
        }
    }

    /**
     * 应用总统计
     */
    public static function statisticProject()
    {
        $transferData = self::$transferData;
        $day = $transferData['day'];
        $project = $transferData['project'];
        $type = $transferData['type'];
        $success = 1 === $transferData['success'];
        $costTime = $transferData['costTime'];
        $key = $transferData['statistic_project_key'];
        if (Redis::exists('key:' . $key)) {
            goto update_db;
        }
        //取锁：有效期3秒，自释放
        $lock = StatisticLocker::lock($key, 3, true);

        //多进程阻塞等待
        if ($lock->acquire(true)) {
            $row = Db::table('statistic_project')->where('key', $key)->first();
            if ($row) {
                update_db:
                //这里不要用查询出的递增字段的值，因为多进程下存在并发，数据会不准！！
                //最佳实践，根据查询出的数据，拼接$where，而$data使用数据库原生语法
                $data = [
                    'count' => Db::raw('count + 1'),
                    'cost'  => Db::raw("cost + {$costTime}"),
                ];
                if ($success) {
                    $data['success_count'] = Db::raw('success_count + 1');
                } else {
                    $data['error_count'] = Db::raw('error_count + 1');
                }
                Db::table('statistic_project')
                    ->where('key', $key)
                    ->update($data);
            } else {
                $data = [
                    'key'           => $key,
                    'day'           => $day,
                    'project'       => $project,
                    'type'          => $type,
                    'count'         => 1,
                    'cost'          => $costTime,
                    'success_count' => $success ? 1 : 0,
                    'error_count'   => $success ? 0 : 1,
                ];
                Db::table('statistic_project')->insert($data);
                //设置标志
                Redis::setEx('key:' . $key, 300, time());
            }
        }
    }

    /**
     * 应用每分钟统计
     */
    public static function statisticProjectInterval()
    {
        $transferData = self::$transferData;
        $day = $transferData['day'];
        $minute = $transferData['minute'];
        $project = $transferData['project'];
        $type = $transferData['type'];
        $success = 1 === $transferData['success'];
        $costTime = $transferData['costTime'];
        $key = $transferData['statistic_project_interval_key'];
        if (Redis::exists('key:' . $key)) {
            goto update_db;
        }
        //取锁：有效期3秒，自释放
        $lock = StatisticLocker::lock($key, 3, true);

        //多进程阻塞等待
        if ($lock->acquire(true)) {
            $row = Db::table('statistic_project_interval')->where('key', $key)->first();
            if ($row) {
                update_db:
                //这里不要用查询出的递增字段的值，因为多进程下存在并发，数据会不准！！
                //最佳实践，根据查询出的数据，拼接$where，而$data使用数据库原生语法
                $data = [
                    'count' => Db::raw('count + 1'),
                    'cost'  => Db::raw("cost + {$costTime}"),
                ];
                if ($success) {
                    $data['success_count'] = Db::raw('success_count + 1');
                } else {
                    $data['error_count'] = Db::raw('error_count + 1');
                }
                Db::table('statistic_project_interval')
                    ->where('key', $key)
                    ->update($data);
            } else {
                $data = [
                    'key'           => $key,
                    'day'           => $day,
                    'time'          => $minute,
                    'project'       => $project,
                    'type'          => $type,
                    'count'         => 1,
                    'cost'          => $costTime,
                    'success_count' => $success ? 1 : 0,
                    'error_count'   => $success ? 0 : 1,
                ];
                Db::table('statistic_project_interval')->insert($data);
                //设置标志
                Redis::setEx('key:' . $key, 300, time());
            }
        }
    }

    /**
     * 应用调用统计
     */
    public static function statisticProjectTransfer()
    {
        $transferData = self::$transferData;
        $day = $transferData['day'];
        $project = $transferData['project'];
        $type = $transferData['type'];
        $costTime = $transferData['costTime'];
        $transfer = $transferData['transfer'];
        $success = 1 === $transferData['success'];
        $key = $transferData['statistic_project_transfer_key'];
        if (Redis::exists('key:' . $key)) {
            goto update_db;
        }
        //取锁：有效期3秒，自释放
        $lock = StatisticLocker::lock($key, 3, true);

        //多进程阻塞等待
        if ($lock->acquire(true)) {
            $row = Db::table('statistic_project_transfer')->where('key', $key)->first();
            if ($row) {
                update_db:
                //这里不要用查询出的递增字段的值，因为多进程下存在并发，数据会不准！！
                //最佳实践，根据查询出的数据，拼接$where，而$data使用数据库原生语法
                $data = [
                    'count' => Db::raw('count + 1'),
                    'cost'  => Db::raw("cost + {$costTime}"),
                ];
                if ($success) {
                    $data['success_count'] = Db::raw('success_count + 1');
                } else {
                    $data['error_count'] = Db::raw('error_count + 1');
                }
                Db::table('statistic_project_transfer')
                    ->where('key', $key)
                    ->update($data);
            } else {
                $data = [
                    'key'           => $key,
                    'day'           => $day,
                    'project'       => $project,
                    'type'          => $type,
                    'transfer'      => $transfer,
                    'count'         => 1,
                    'cost'          => $costTime,
                    'success_count' => $success ? 1 : 0,
                    'error_count'   => $success ? 0 : 1,
                ];
                Db::table('statistic_project_transfer')->insert($data);
                //设置标志
                Redis::setEx('key:' . $key, 300, time());
            }
        }
    }

    /**
     * 应用调用每分钟统计
     */
    public static function statisticProjectTransferInterval()
    {
        $transferData = self::$transferData;
        $day = $transferData['day'];
        $minute = $transferData['minute'];
        $project = $transferData['project'];
        $type = $transferData['type'];
        $costTime = $transferData['costTime'];
        $transfer = $transferData['transfer'];
        $success = 1 === $transferData['success'];
        $key = $transferData['statistic_project_transfer_interval_key'];
        if (Redis::exists('key:' . $key)) {
            goto update_db;
        }
        //取锁：有效期3秒，自释放
        $lock = StatisticLocker::lock($key, 3, true);

        //多进程阻塞等待
        if ($lock->acquire(true)) {
            $row = Db::table('statistic_project_transfer_interval')->where('key', $key)->first();
            if ($row) {
                update_db:
                //这里不要用查询出的递增字段的值，因为多进程下存在并发，数据会不准！！
                //最佳实践，根据查询出的数据，拼接$where，而$data使用数据库原生语法
                $data = [
                    'count' => Db::raw('count + 1'),
                    'cost'  => Db::raw("cost + {$costTime}"),
                ];
                if ($success) {
                    $data['success_count'] = Db::raw('success_count + 1');
                } else {
                    $data['error_count'] = Db::raw('error_count + 1');
                }
                Db::table('statistic_project_transfer_interval')
                    ->where('key', $key)
                    ->update($data);
            } else {
                $data = [
                    'key'           => $key,
                    'day'           => $day,
                    'time'          => $minute,
                    'project'       => $project,
                    'type'          => $type,
                    'transfer'      => $transfer,
                    'count'         => 1,
                    'cost'          => $costTime,
                    'success_count' => $success ? 1 : 0,
                    'error_count'   => $success ? 0 : 1,
                ];
                Db::table('statistic_project_transfer_interval')->insert($data);
                //设置标志
                Redis::setEx('key:' . $key, 300, time());
            }
        }
    }

    /**
     * 应用状态码统计
     */
    public static function statisticProjectCode()
    {
        $transferData = self::$transferData;
        $day = $transferData['day'];
        $project = $transferData['project'];
        $type = $transferData['type'];
        $costTime = $transferData['costTime'];
        $code = $transferData['code'];
        $key = $transferData['statistic_project_code_key'];
        if (Redis::exists('key:' . $key)) {
            goto update_db;
        }
        //取锁：有效期3秒，自释放
        $lock = StatisticLocker::lock($key, 3, true);

        //多进程阻塞等待
        if ($lock->acquire(true)) {
            $row = Db::table('statistic_project_code')->where('key', $key)->first();
            if ($row) {
                update_db:
                //这里不要用查询出的递增字段的值，因为多进程下存在并发，数据会不准！！
                //最佳实践，根据查询出的数据，拼接$where，而$data使用数据库原生语法
                Db::table('statistic_project_code')
                    ->where('key', $key)
                    ->update([
                        'count' => Db::raw('count + 1'),
                        'cost'  => Db::raw("cost + {$costTime}"),
                    ]);
            } else {
                $data = [
                    'key'     => $key,
                    'day'     => $day,
                    'project' => $project,
                    'type'    => $type,
                    'code'    => $code,
                    'count'   => 1,
                    'cost'    => $costTime,
                ];
                Db::table('statistic_project_code')->insert($data);
                //设置标志
                Redis::setEx('key:' . $key, 300, time());
            }
        }
    }

    /**
     * 应用状态码每分钟统计
     */
    public static function statisticProjectCodeInterval()
    {
        $transferData = self::$transferData;
        $day = $transferData['day'];
        $minute = $transferData['minute'];
        $project = $transferData['project'];
        $type = $transferData['type'];
        $costTime = $transferData['costTime'];
        $code = $transferData['code'];
        $key = $transferData['statistic_project_code_interval_key'];
        if (Redis::exists('key:' . $key)) {
            goto update_db;
        }
        //取锁：有效期3秒，自释放
        $lock = StatisticLocker::lock($key, 3, true);

        //多进程阻塞等待
        if ($lock->acquire(true)) {
            $row = Db::table('statistic_project_code_interval')->where('key', $key)->first();
            if ($row) {
                update_db:
                //这里不要用查询出的递增字段的值，因为多进程下存在并发，数据会不准！！
                //最佳实践，根据查询出的数据，拼接$where，而$data使用数据库原生语法
                Db::table('statistic_project_code_interval')
                    ->where('key', $key)
                    ->update([
                        'count' => Db::raw('count + 1'),
                        'cost'  => Db::raw("cost + {$costTime}"),
                    ]);
            } else {
                $data = [
                    'key'     => $key,
                    'day'     => $day,
                    'time'    => $minute,
                    'project' => $project,
                    'type'    => $type,
                    'code'    => $code,
                    'count'   => 1,
                    'cost'    => $costTime,
                ];
                Db::table('statistic_project_code_interval')->insert($data);
                //设置标志
                Redis::setEx('key:' . $key, 300, time());
            }
        }
    }

    /**
     * 应用IP统计
     */
    public static function statisticProjectIp()
    {
        $transferData = self::$transferData;
        $day = $transferData['day'];
        $project = $transferData['project'];
        $type = $transferData['type'];
        $costTime = $transferData['costTime'];
        $success = 1 === $transferData['success'];
        $ip = $transferData['ip'];
        $key = $transferData['statistic_project_ip_key'];
        if (Redis::exists('key:' . $key)) {
            goto update_db;
        }
        //取锁：有效期3秒，自释放
        $lock = StatisticLocker::lock($key, 3, true);

        //多进程阻塞等待
        if ($lock->acquire(true)) {
            $row = Db::table('statistic_project_ip')->where('key', $key)->first();
            if ($row) {
                update_db:
                //这里不要用查询出的递增字段的值，因为多进程下存在并发，数据会不准！！
                //最佳实践，根据查询出的数据，拼接$where，而$data使用数据库原生语法
                $data = [
                    'count' => Db::raw('count + 1'),
                    'cost'  => Db::raw("cost + {$costTime}"),
                ];
                if ($success) {
                    $data['success_count'] = Db::raw('success_count + 1');
                } else {
                    $data['error_count'] = Db::raw('error_count + 1');
                }
                Db::table('statistic_project_ip')
                    ->where('key', $key)
                    ->update($data);
            } else {
                $data = [
                    'key'           => $key,
                    'day'           => $day,
                    'project'       => $project,
                    'type'          => $type,
                    'ip'            => $ip,
                    'count'         => 1,
                    'cost'          => $costTime,
                    'success_count' => $success ? 1 : 0,
                    'error_count'   => $success ? 0 : 1,
                ];
                Db::table('statistic_project_ip')->insert($data);
                //设置标志
                Redis::setEx('key:' . $key, 300, time());
            }
        }
    }

    /**
     * 应用IP每分钟统计
     */
    public static function statisticProjectIpInterval()
    {
        $transferData = self::$transferData;
        $day = $transferData['day'];
        $minute = $transferData['minute'];
        $project = $transferData['project'];
        $type = $transferData['type'];
        $costTime = $transferData['costTime'];
        $success = 1 === $transferData['success'];
        $ip = $transferData['ip'];
        $key = $transferData['statistic_project_ip_interval_key'];
        if (Redis::exists('key:' . $key)) {
            goto update_db;
        }
        //取锁：有效期3秒，自释放
        $lock = StatisticLocker::lock($key, 3, true);

        //多进程阻塞等待
        if ($lock->acquire(true)) {
            $row = Db::table('statistic_project_ip_interval')->where('key', $key)->first();
            if ($row) {
                update_db:
                //这里不要用查询出的递增字段的值，因为多进程下存在并发，数据会不准！！
                //最佳实践，根据查询出的数据，拼接$where，而$data使用数据库原生语法
                $data = [
                    'count' => Db::raw('count + 1'),
                    'cost'  => Db::raw("cost + {$costTime}"),
                ];
                if ($success) {
                    $data['success_count'] = Db::raw('success_count + 1');
                } else {
                    $data['error_count'] = Db::raw('error_count + 1');
                }
                Db::table('statistic_project_ip_interval')
                    ->where('key', $key)
                    ->update($data);
            } else {
                $data = [
                    'key'           => $key,
                    'day'           => $day,
                    'time'          => $minute,
                    'project'       => $project,
                    'type'          => $type,
                    'ip'            => $ip,
                    'count'         => 1,
                    'cost'          => $costTime,
                    'success_count' => $success ? 1 : 0,
                    'error_count'   => $success ? 0 : 1,
                ];
                Db::table('statistic_project_ip_interval')->insert($data);
                //设置标志
                Redis::setEx('key:' . $key, 300, time());
            }
        }
    }

    /**
     * 应用IP状态码统计
     */
    public static function statisticProjectIpCode()
    {
        $transferData = self::$transferData;
        $day = $transferData['day'];
        $project = $transferData['project'];
        $type = $transferData['type'];
        $costTime = $transferData['costTime'];
        $ip = $transferData['ip'];
        $code = $transferData['code'];
        $key = $transferData['statistic_project_ip_code_key'];
        if (Redis::exists('key:' . $key)) {
            goto update_db;
        }
        //取锁：有效期3秒，自释放
        $lock = StatisticLocker::lock($key, 3, true);

        //多进程阻塞等待
        if ($lock->acquire(true)) {
            $row = Db::table('statistic_project_ip_code')->where('key', $key)->first();
            if ($row) {
                update_db:
                //这里不要用查询出的递增字段的值，因为多进程下存在并发，数据会不准！！
                //最佳实践，根据查询出的数据，拼接$where，而$data使用数据库原生语法

                Db::table('statistic_project_ip_code')
                    ->where('key', $key)
                    ->update([
                        'count' => Db::raw('count + 1'),
                        'cost'  => Db::raw("cost + {$costTime}"),
                    ]);
            } else {
                $data = [
                    'key'     => $key,
                    'day'     => $day,
                    'project' => $project,
                    'type'    => $type,
                    'ip'      => $ip,
                    'code'    => $code,
                    'count'   => 1,
                    'cost'    => $costTime,
                ];
                Db::table('statistic_project_ip_code')->insert($data);
                //设置标志
                Redis::setEx('key:' . $key, 300, time());
            }
        }
    }

    /**
     * 应用IP状态码每分钟统计
     */
    public static function statisticProjectIpCodeInterval()
    {
        $transferData = self::$transferData;
        $day = $transferData['day'];
        $minute = $transferData['minute'];
        $project = $transferData['project'];
        $type = $transferData['type'];
        $costTime = $transferData['costTime'];
        $ip = $transferData['ip'];
        $code = $transferData['code'];
        $key = $transferData['statistic_project_ip_code_interval_key'];
        if (Redis::exists('key:' . $key)) {
            goto update_db;
        }
        //取锁：有效期3秒，自释放
        $lock = StatisticLocker::lock($key, 3, true);

        //多进程阻塞等待
        if ($lock->acquire(true)) {
            $row = Db::table('statistic_project_ip_code_interval')->where('key', $key)->first();
            if ($row) {
                update_db:
                //这里不要用查询出的递增字段的值，因为多进程下存在并发，数据会不准！！
                //最佳实践，根据查询出的数据，拼接$where，而$data使用数据库原生语法
                Db::table('statistic_project_ip_code_interval')
                    ->where('key', $key)
                    ->update([
                        'count' => Db::raw('count + 1'),
                        'cost'  => Db::raw("cost + {$costTime}"),
                    ]);
            } else {
                $data = [
                    'key'     => $key,
                    'day'     => $day,
                    'time'    => $minute,
                    'project' => $project,
                    'type'    => $type,
                    'ip'      => $ip,
                    'code'    => $code,
                    'count'   => 1,
                    'cost'    => $costTime,
                ];
                Db::table('statistic_project_ip_code_interval')->insert($data);
                //设置标志
                Redis::setEx('key:' . $key, 300, time());
            }
        }
    }

    /**
     * 应用IP调用统计
     */
    public static function statisticProjectIpTransfer()
    {
        $transferData = self::$transferData;
        $day = $transferData['day'];
        $project = $transferData['project'];
        $type = $transferData['type'];
        $costTime = $transferData['costTime'];
        $ip = $transferData['ip'];
        $transfer = $transferData['transfer'];
        $success = 1 === $transferData['success'];
        $key = $transferData['statistic_project_ip_transfer_key'];
        if (Redis::exists('key:' . $key)) {
            goto update_db;
        }
        //取锁：有效期3秒，自释放
        $lock = StatisticLocker::lock($key, 3, true);

        //多进程阻塞等待
        if ($lock->acquire(true)) {
            $row = Db::table('statistic_project_ip_transfer')->where('key', $key)->first();
            if ($row) {
                update_db:
                //这里不要用查询出的递增字段的值，因为多进程下存在并发，数据会不准！！
                //最佳实践，根据查询出的数据，拼接$where，而$data使用数据库原生语法

                $data = [
                    'count' => Db::raw('count + 1'),
                    'cost'  => Db::raw("cost + {$costTime}"),
                ];
                if ($success) {
                    $data['success_count'] = Db::raw('success_count + 1');
                } else {
                    $data['error_count'] = Db::raw('error_count + 1');
                }
                Db::table('statistic_project_ip_transfer')
                    ->where('key', $key)
                    ->update($data);
            } else {
                $data = [
                    'key'           => $key,
                    'day'           => $day,
                    'project'       => $project,
                    'type'          => $type,
                    'ip'            => $ip,
                    'transfer'      => $transfer,
                    'count'         => 1,
                    'cost'          => $costTime,
                    'success_count' => $success ? 1 : 0,
                    'error_count'   => $success ? 0 : 1,
                ];
                Db::table('statistic_project_ip_transfer')->insert($data);
                //设置标志
                Redis::setEx('key:' . $key, 300, time());
            }
        }
    }

    /**
     * 应用IP调用每分钟统计
     */
    public static function statisticProjectIpTransferInterval()
    {
        $transferData = self::$transferData;
        $day = $transferData['day'];
        $minute = $transferData['minute'];
        $project = $transferData['project'];
        $type = $transferData['type'];
        $costTime = $transferData['costTime'];
        $ip = $transferData['ip'];
        $transfer = $transferData['transfer'];
        $success = 1 === $transferData['success'];
        $key = $transferData['statistic_project_ip_transfer_interval_key'];

        if (Redis::exists('key:' . $key)) {
            goto update_db;
        }
        //取锁：有效期3秒，自释放
        $lock = StatisticLocker::lock($key, 3, true);

        //多进程阻塞等待
        if ($lock->acquire(true)) {
            $row = Db::table('statistic_project_ip_transfer_interval')->where('key', $key)->first();
            if ($row) {
                update_db:
                //这里不要用查询出的递增字段的值，因为多进程下存在并发，数据会不准！！
                //最佳实践，根据查询出的数据，拼接$where，而$data使用数据库原生语法
                $data = [
                    'count' => Db::raw('count + 1'),
                    'cost'  => Db::raw("cost + {$costTime}"),
                ];
                if ($success) {
                    $data['success_count'] = Db::raw('success_count + 1');
                } else {
                    $data['error_count'] = Db::raw('error_count + 1');
                }
                Db::table('statistic_project_ip_transfer_interval')
                    ->where('key', $key)
                    ->update($data);
            } else {
                $data = [
                    'key'           => $key,
                    'day'           => $day,
                    'time'          => $minute,
                    'project'       => $project,
                    'type'          => $type,
                    'ip'            => $ip,
                    'transfer'      => $transfer,
                    'count'         => 1,
                    'cost'          => $costTime,
                    'success_count' => $success ? 1 : 0,
                    'error_count'   => $success ? 0 : 1,
                ];
                Db::table('statistic_project_ip_transfer_interval')->insert($data);
                //设置标志
                Redis::setEx('key:' . $key, 300, time());
            }
        }
    }
}

上述代码经过测试，300条数据，单进程耗时20.487082958221436s，cpu数进程耗时21.118263483047485s，cpu*4进程数耗时22.74993324279785s，对进程处理并没有加快反而由于锁的存在多耗时了；大佬帮看看我上面的代码，我准备放弃了，改用redis+定时器扫描批量入库了

//取锁：有效期3秒，自释放
$lock = StatisticLocker::lock($key, 3, true);

数据库频繁的插入更新可能导致每次锁的释放时间增加；

根据业务逻辑，insert只有一次，update可能存在很多次，这一块是造成数据库压力的原因，所以建议把update部分的内容放到redis队列，其他逻辑保持不变。

那个锁插件的基类代码：

class Locker
{
    public static function __callStatic($name, $arguments)
    {
        $key = $arguments[0] ?? '';
        unset($arguments[0]);
        return static::createLock($name . $key, ...$arguments);
    }

而你用的都是这一个方法，key的取值是否有重复，不得而知。

//取锁：有效期3秒，自释放
$lock = StatisticLocker::lock($key, 3, true);

每一个表的操作是不能共享锁的。也要根据业务的特性，ttl值也是有讲究的。
比如每天、每分钟、每小时才创建一次的，要根据时间合理调整ttl值。
总体原则是：创建时候才加锁，多进程同步更新。
我记得我给你写了很多个方法的啊？为的就是避免锁重复

每一个方法名，表示一个锁的前缀，前缀也是锁key的一部分（可看源码）；
意思就是前缀不同、key值相同，也是不同的锁。

/**
 * 统计锁
 * @method static LockInterface lock(string $day, ?float $ttl = null, ?bool $autoRelease = null) 数据库操作锁
 * @method static LockInterface project(string $project, ?float $ttl = null, ?bool $autoRelease = null) 应用锁
 * @method static LockInterface statistic(string $day, ?float $ttl = null, ?bool $autoRelease = null) 总统计锁
 * @method static LockInterface statisticProject(string $day_project_type, ?float $ttl = null, ?bool $autoRelease = null) 应用总统计
 * @method static LockInterface statisticProjectInterval(string $day_minute_project_type, ?float $ttl = null, ?bool $autoRelease = null) 应用每分钟统计
 * @method static LockInterface statisticProjectTransfer(string $key, ?float $ttl = null, ?bool $autoRelease = null) 应用调用统计
 * @method static LockInterface statisticProjectTransferInterval(string $key, ?float $ttl = null, ?bool $autoRelease = null) 应用调用每分钟统计
 * @method static LockInterface statisticProjectCode(string $key, ?float $ttl = null, ?bool $autoRelease = null) 应用状态码统计
 * @method static LockInterface statisticProjectCodeInterval(string $key, ?float $ttl = null, ?bool $autoRelease = null) 应用状态码每分钟统计
 * @method static LockInterface statisticProjectIp(string $key, ?float $ttl = null, ?bool $autoRelease = null) 应用IP统计
 * @method static LockInterface statisticProjectIpInterval(string $key, ?float $ttl = null, ?bool $autoRelease = null) 应用IP每分钟统计
 * @method static LockInterface statisticProjectIpCode(string $key, ?float $ttl = null, ?bool $autoRelease = null) 应用IP状态码统计
 * @method static LockInterface statisticProjectIpCodeInterval(string $key, ?float $ttl = null, ?bool $autoRelease = null) 应用IP状态码每分钟统计
 * @method static LockInterface statisticProjectIpTransfer(string $key, ?float $ttl = null, ?bool $autoRelease = null) 应用IP调用统计
 * @method static LockInterface statisticProjectIpTransferInterval(string $key, ?float $ttl = null, ?bool $autoRelease = null) 应用IP调用每分钟统计
 */
class StatisticLocker extends Locker
{

}

我们之前访问量不大，直接是使用事务，保证数据的一致性，我问了ai，它让我使用行锁

use Illuminate\Support\Facades\DB;

// 开启事务
DB::beginTransaction();

try {
    // 使用行级锁查询数据
    $row = DB::table('statistic')->where('day', $day)->lockForUpdate()->first();

    if ($row) {
        $data = [
            'count'         => $row->count + 1,
            'cost'          => $row->cost + $costTime,
            'success_count' => $row->success_count + ($success ? 1 : 0),
            'error_count'   => $row->error_count + ($success ? 0 : 1),
        ];
        DB::table('statistic')->where('day', $day)->update($data);
    } else {
        $data = [
            'day'           => $day,
            'count'         => 1,
            'cost'          => $costTime,
            'success_count' => $success ? 1 : 0,
            'error_count'   => $success ? 0 : 1,
        ];
        DB::table('statistic')->insert($data);
    }

    // 提交事务
    DB::commit();
} catch (\Exception $e) {
    // 回滚事务
    DB::rollback();
    // 处理异常
}

队列消费不是应该消出库就没了，不存在二次消费吧

你这种情况是多进程并发所致，你的SQL语句并非原子操作。
想要解决这个问题，你只能使用MySQL 自带SQL语句来实现原子操作。
这篇文章能解决这些问题，https://www.ngui.cc/el/1890018.html?action=onClick

建议采用这条SQL写法：ON DUPLICATE KEY UPDATE