yueyang
cbbf32bb5c
映射表保存、删除、恢复、导入和重置后，自动把受影响平台已存在的月份排入后台队列；后台按平台×月份逐个独立进程重跑，数据库锁防重复，生产 cron 每分钟兜底。保存响应直接提示排队任务数。
108 lines
4.5 KiB
PHP
108 lines
4.5 KiB
PHP
<?php
|
||
|
||
namespace App\Console\Commands;
|
||
|
||
use Illuminate\Console\Command;
|
||
use Illuminate\Support\Facades\Cache;
|
||
use Illuminate\Support\Facades\DB;
|
||
use Illuminate\Support\Facades\Log;
|
||
use Modules\EcommerceFinance\Services\AccountMappingResyncDispatcher;
|
||
use Symfony\Component\Process\Process;
|
||
use Throwable;
|
||
|
||
/**
|
||
* 科目映射表变更后的业绩明细自动重跑队列。
|
||
*
|
||
* 一个常驻进程依次处理队列里的「平台 × 月份」；每个实际同步放在独立 PHP 子进程中，
|
||
* 避免大平台重跑时的内存累积。也可手动执行本命令补跑。
|
||
*/
|
||
class RunAccountMappingResync extends Command
|
||
{
|
||
protected $signature = 'store-receipt:auto-resync';
|
||
|
||
protected $description = '处理科目映射表变更触发的业绩明细重跑队列';
|
||
|
||
public function handle(AccountMappingResyncDispatcher $dispatcher): int
|
||
{
|
||
// 页面（Apache）与每分钟 cron 都以 www-data 运行，使用共享数据库锁，
|
||
// 不受 storage/framework/cache 文件属主影响，也能避免多容器重复消费。
|
||
$lock = Cache::store('upstream_shared')->lock('store-receipt:auto-resync', 86400);
|
||
if (! $lock->get()) {
|
||
// 已有进程在跑；本次新入队的任务会在当前进程下一轮循环或每分钟兜底时执行。
|
||
return self::SUCCESS;
|
||
}
|
||
|
||
try {
|
||
// 进程崩溃遗留的 running 任务超过 60 分钟视为可重试。
|
||
DB::table(AccountMappingResyncDispatcher::JOBS_TABLE)
|
||
->where('status', 'running')
|
||
->where('updated_at', '<', now()->subMinutes(60))
|
||
->update(['status' => 'pending', 'updated_at' => now()]);
|
||
|
||
while (true) {
|
||
$job = DB::table(AccountMappingResyncDispatcher::JOBS_TABLE)
|
||
->where('status', 'pending')
|
||
->orderBy('id')
|
||
->first();
|
||
if ($job === null) {
|
||
break;
|
||
}
|
||
|
||
DB::table(AccountMappingResyncDispatcher::JOBS_TABLE)
|
||
->where('id', $job->id)
|
||
->where('status', 'pending')
|
||
->update([
|
||
'status' => 'running',
|
||
'attempts' => DB::raw('attempts + 1'),
|
||
'updated_at' => now(),
|
||
]);
|
||
|
||
try {
|
||
$process = Process::fromShellCommandline(
|
||
implode(' ', [
|
||
escapeshellarg(PHP_BINARY ?: 'php'),
|
||
'artisan',
|
||
'store-receipt:order-match:sync',
|
||
escapeshellarg((string) $job->month),
|
||
'--platform=' . escapeshellarg((string) $job->platform),
|
||
]),
|
||
base_path()
|
||
);
|
||
$process->setTimeout(7200);
|
||
$process->run();
|
||
$output = trim($process->getOutput() . "\n" . $process->getErrorOutput());
|
||
|
||
// 同步期间又保存了映射时，dispatchFor 已把状态改回 pending；
|
||
// 这里只在仍为 running 时写 done，不能吞掉新请求。
|
||
DB::table(AccountMappingResyncDispatcher::JOBS_TABLE)
|
||
->where('id', $job->id)
|
||
->where('status', 'running')
|
||
->update([
|
||
'status' => $process->isSuccessful() ? 'done' : 'failed',
|
||
'message' => mb_substr(
|
||
$output !== '' ? $output : ($process->isSuccessful() ? 'OK' : '同步命令执行失败'),
|
||
0,
|
||
1000
|
||
),
|
||
'updated_at' => now(),
|
||
]);
|
||
} catch (Throwable $exception) {
|
||
Log::error('科目映射表自动重跑失败：' . $exception->getMessage());
|
||
DB::table(AccountMappingResyncDispatcher::JOBS_TABLE)
|
||
->where('id', $job->id)
|
||
->where('status', 'running')
|
||
->update([
|
||
'status' => 'failed',
|
||
'message' => mb_substr($exception->getMessage(), 0, 1000),
|
||
'updated_at' => now(),
|
||
]);
|
||
}
|
||
}
|
||
} finally {
|
||
$lock->release();
|
||
}
|
||
|
||
return self::SUCCESS;
|
||
}
|
||
}
|