107 lines
3.6 KiB
PHP
107 lines
3.6 KiB
PHP
<?php
|
||
|
||
namespace App\Console\Commands;
|
||
|
||
use App\Blog\Services\LlmClient;
|
||
use Illuminate\Console\Command;
|
||
use Illuminate\Support\Facades\Artisan;
|
||
use Illuminate\Support\Facades\DB;
|
||
use Illuminate\Support\Facades\Log;
|
||
use Workerman\Connection\TcpConnection;
|
||
use Workerman\Timer;
|
||
use Workerman\Worker;
|
||
|
||
class WorkermanServe extends Command
|
||
{
|
||
protected $signature = 'workerman:serve {action=start : start/stop/restart/reload}';
|
||
|
||
protected $description = '启动 Workerman 常驻服务：队列消费者（LLM 异步任务）+ WebSocket 进度推送';
|
||
|
||
public function handle(): int
|
||
{
|
||
Worker::$pidFile = storage_path('framework/workerman.pid');
|
||
|
||
$this->info('Workerman 启动中（'.config('workerman.name').'）...');
|
||
|
||
// 队列消费者：常驻进程内复用 Laravel 容器，避免每次任务重新启动框架
|
||
$queueWorker = new Worker();
|
||
$queueWorker->name = 'laralog-queue';
|
||
$queueWorker->count = (int) config('workerman.queue_workers', 2);
|
||
$queueWorker->onWorkerStart = function ($worker) {
|
||
$this->info("队列消费者 {$worker->id} 启动");
|
||
Timer::add(1, function () {
|
||
try {
|
||
$this->consumeNextJob();
|
||
} catch (\Throwable $e) {
|
||
Log::error('Workerman 任务执行失败', ['error' => $e->getMessage()]);
|
||
}
|
||
});
|
||
};
|
||
|
||
// WebSocket：向后台推送任务进度
|
||
$ws = new Worker('websocket://'.config('workerman.websocket_host', '0.0.0.0').':'.config('workerman.websocket_port', 8787));
|
||
$ws->name = 'laralog-ws';
|
||
$ws->onConnect = function (TcpConnection $connection) {
|
||
\App\Blog\Support\WorkermanBroadcaster::add($connection);
|
||
$connection->send(json_encode(['event' => 'connected']));
|
||
};
|
||
$ws->onClose = fn (TcpConnection $connection) => \App\Blog\Support\WorkermanBroadcaster::remove($connection);
|
||
|
||
Worker::runAll();
|
||
|
||
return self::SUCCESS;
|
||
}
|
||
|
||
private function consumeNextJob(): void
|
||
{
|
||
$queue = config('workerman.queue_connection', 'database');
|
||
|
||
if ($queue === 'database') {
|
||
$job = DB::table('jobs')
|
||
->whereNull('reserved_at')
|
||
->orderBy('id')
|
||
->lockForUpdate()
|
||
->first();
|
||
|
||
if (! $job) {
|
||
return;
|
||
}
|
||
|
||
DB::table('jobs')->where('id', $job->id)->update([
|
||
'reserved_at' => now()->getTimestamp(),
|
||
'attempts' => $job->attempts + 1,
|
||
]);
|
||
|
||
$payload = json_decode($job->payload, true);
|
||
|
||
try {
|
||
$this->runJob($payload);
|
||
DB::table('jobs')->where('id', $job->id)->delete();
|
||
} catch (\Throwable $e) {
|
||
Log::error('任务失败', ['job' => $job->id, 'error' => $e->getMessage()]);
|
||
|
||
$attempts = $job->attempts + 1;
|
||
if ($attempts >= 3) {
|
||
DB::table('jobs')->where('id', $job->id)->update(['reserved_at' => null, 'attempts' => $attempts]);
|
||
} else {
|
||
DB::table('jobs')->where('id', $job->id)->delete();
|
||
}
|
||
}
|
||
|
||
return;
|
||
}
|
||
|
||
// Redis 队列：降级到 artisan queue:work
|
||
Artisan::call('queue:work', ['--once' => true, '--stop-when-empty' => true]);
|
||
}
|
||
|
||
private function runJob(array $payload): void
|
||
{
|
||
$command = unserialize($payload['data']['command'] ?? '');
|
||
|
||
if ($command instanceof \App\Blog\Jobs\AiJob) {
|
||
$command->handle(app(LlmClient::class));
|
||
}
|
||
}
|
||
}
|