diff --git a/.env.example b/.env.example index c0660ea..36e969e 100644 --- a/.env.example +++ b/.env.example @@ -63,3 +63,14 @@ AWS_BUCKET= AWS_USE_PATH_STYLE_ENDPOINT=false VITE_APP_NAME="${APP_NAME}" + +# ---- Workerman 常驻服务 ---- +WORKERMAN_QUEUE_WORKERS=2 +WORKERMAN_WS_ENABLED=true +WORKERMAN_WS_PORT=8787 +# 队列连接：database / redis（redis 需要先配置好下方 REDIS_* 与 QUEUE_CONNECTION=redis） +WORKERMAN_QUEUE_CONNECTION=database +# 消费的队列名，逗号分隔；ai 队列为 AI 审核/润色任务 +WORKERMAN_QUEUES=default,ai +WORKERMAN_MAX_TRIES=3 +WORKERMAN_TIMEOUT=60 diff --git a/README.md b/README.md index 089719b..146a1a3 100644 --- a/README.md +++ b/README.md @@ -92,8 +92,13 @@ php artisan workerman:serve start # 队列消费者 × N + WebSocket :8787 php artisan workerman:serve stop ``` -- 队列消费者在常驻进程内复用 Laravel 容器，省去每次任务的框架启动开销（LLM 调用主路径） -- 任务统一实现 `App\Blog\Jobs\AiJob` 接口（`handle(LlmClient)`），Workerman 反序列化执行 +- 消费者复用 Laravel `Queue\Worker`，**database / redis 队列统一支持**，常驻进程内框架只启动一次（省去每个任务的启动开销） +- 任务统一实现 `App\Blog\Jobs\AiJob` 接口（`handle(LlmClient)`），依赖由容器自动注入 +- 队列选择： + - database（默认）：`WORKERMAN_QUEUE_CONNECTION=database` + - redis：`WORKERMAN_QUEUE_CONNECTION=redis` 并确保 `QUEUE_CONNECTION=redis`、`REDIS_*` 配置正确；`redis` 连接的 `block_for` 建议设为 `0`（配合 1s 轮询，避免阻塞 Workerman event loop） +- 消费队列：`WORKERMAN_QUEUES=default,ai`（ai 队列为 AI 审核/润色任务） +- 失败重试：`WORKERMAN_MAX_TRIES`（默认 3），超时 `WORKERMAN_TIMEOUT`（默认 60s），重试耗尽进 `failed_jobs` 表 - 降级路径：`php artisan queue:work` 照常可用（同一队列） ## 测试 diff --git a/app/Console/Commands/WorkermanServe.php b/app/Console/Commands/WorkermanServe.php index 9cd767d..7eb2829 100644 --- a/app/Console/Commands/WorkermanServe.php +++ b/app/Console/Commands/WorkermanServe.php @@ -2,10 +2,10 @@ 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\Contracts\Debug\ExceptionHandler; +use Illuminate\Queue\Worker as QueueWorker; +use Illuminate\Queue\WorkerOptions; use Illuminate\Support\Facades\Log; use Workerman\Connection\TcpConnection; use Workerman\Timer; @@ -15,7 +15,7 @@ class WorkermanServe extends Command { protected $signature = 'workerman:serve {action=start : start/stop/restart/reload}'; - protected $description = '启动 Workerman 常驻服务：队列消费者（LLM 异步任务）+ WebSocket 进度推送'; + protected $description = '启动 Workerman 常驻服务：队列消费者（database/redis）+ WebSocket 进度推送'; public function handle(): int { @@ -23,12 +23,13 @@ class WorkermanServe extends Command $this->info('Workerman 启动中（'.config('workerman.name').'）...'); - // 队列消费者：常驻进程内复用 Laravel 容器，避免每次任务重新启动框架 + // 队列消费者：复用 Laravel 队列 Worker（database / redis 统一支持）， + // 常驻进程内框架只启动一次，省去每个任务的启动开销 $queueWorker = new Worker(); $queueWorker->name = 'laralog-queue'; $queueWorker->count = (int) config('workerman.queue_workers', 2); $queueWorker->onWorkerStart = function ($worker) { - $this->info("队列消费者 {$worker->id} 启动"); + $this->info("队列消费者 {$worker->id} 启动（连接：".config('workerman.queue_connection').'，队列：'.config('workerman.queues').'）'); Timer::add(1, function () { try { $this->consumeNextJob(); @@ -52,55 +53,35 @@ class WorkermanServe extends Command return self::SUCCESS; } + /** + * 消费一个任务。使用 Laravel 队列 Worker： + * - 支持 database / redis 等所有驱动（redis 的 reserved/重试/超时语义由驱动处理） + * - sleep=0 避免阻塞 Workerman event loop（由 1s Timer 驱动轮询） + * - 任务 handle() 的依赖（如 LlmClient）由容器自动注入 + */ private function consumeNextJob(): void { - $queue = config('workerman.queue_connection', 'database'); + $connection = (string) config('workerman.queue_connection', config('queue.default')); + // Worker 内部按逗号分隔解析多队列（如 "default,ai"） + $queues = (string) config('workerman.queues', 'default'); + $queues = $queues === '' ? 'default' : $queues; - if ($queue === 'database') { - $job = DB::table('jobs') - ->whereNull('reserved_at') - ->orderBy('id') - ->lockForUpdate() - ->first(); + $worker = new QueueWorker( + app('queue'), + app('events'), + app(ExceptionHandler::class), + fn () => app()->isDownForMaintenance(), + ); - if (! $job) { - return; - } + $options = new WorkerOptions( + name: 'laralog', + backoff: 0, + memory: 128, + timeout: (int) config('workerman.timeout', 60), + sleep: 0, + maxTries: (int) config('workerman.max_tries', 3), + ); - 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)); - } + $worker->runNextJob($connection, $queues, $options); } } diff --git a/config/workerman.php b/config/workerman.php index 158c097..7d1e42e 100644 --- a/config/workerman.php +++ b/config/workerman.php @@ -22,12 +22,24 @@ return [ /* |-------------------------------------------------------------------------- - | 队列：Workerman 消费者使用的队列连接 + | 队列：Workerman 消费者使用的队列连接（database / redis 均可） |-------------------------------------------------------------------------- */ 'queue_connection' => env('WORKERMAN_QUEUE_CONNECTION', env('QUEUE_CONNECTION', 'database')), + /* + |-------------------------------------------------------------------------- + | 消费的队列名（逗号分隔；ai 队列为 AI 审核/润色任务） + |-------------------------------------------------------------------------- + */ + + 'queues' => env('WORKERMAN_QUEUES', 'default,ai'), + + 'max_tries' => env('WORKERMAN_MAX_TRIES', 3), + + 'timeout' => env('WORKERMAN_TIMEOUT', 60), + /* |-------------------------------------------------------------------------- | 进程名称（ps 可见） diff --git a/storage/media-library/temp/IW4cbSzBhuKZ5WzWL7eOutZtR5bhoiN2/0hI0aMNDcCrIWUDSN8ybSS8W13NbCJqtthumb.jpg b/storage/media-library/temp/IW4cbSzBhuKZ5WzWL7eOutZtR5bhoiN2/0hI0aMNDcCrIWUDSN8ybSS8W13NbCJqtthumb.jpg new file mode 100644 index 0000000..7e754f4 --- /dev/null +++ b/storage/media-library/temp/IW4cbSzBhuKZ5WzWL7eOutZtR5bhoiN2/0hI0aMNDcCrIWUDSN8ybSS8W13NbCJqtthumb.jpg @@ -0,0 +1 @@ +fake-jpg \ No newline at end of file diff --git a/storage/media-library/temp/IW4cbSzBhuKZ5WzWL7eOutZtR5bhoiN2/9OSfeP65XwvrIbbAV0SL6MJt5omEVSAM.jpg b/storage/media-library/temp/IW4cbSzBhuKZ5WzWL7eOutZtR5bhoiN2/9OSfeP65XwvrIbbAV0SL6MJt5omEVSAM.jpg new file mode 100644 index 0000000..7e754f4 --- /dev/null +++ b/storage/media-library/temp/IW4cbSzBhuKZ5WzWL7eOutZtR5bhoiN2/9OSfeP65XwvrIbbAV0SL6MJt5omEVSAM.jpg @@ -0,0 +1 @@ +fake-jpg \ No newline at end of file diff --git a/storage/media-library/temp/LZvUnQohKyX7zYlAWEVEV3OmmaBLNv9I/Y9rDVBYrWxqGPdSUuSk5JE3BkvgK2yRu.jpg b/storage/media-library/temp/LZvUnQohKyX7zYlAWEVEV3OmmaBLNv9I/Y9rDVBYrWxqGPdSUuSk5JE3BkvgK2yRu.jpg new file mode 100644 index 0000000..7e754f4 --- /dev/null +++ b/storage/media-library/temp/LZvUnQohKyX7zYlAWEVEV3OmmaBLNv9I/Y9rDVBYrWxqGPdSUuSk5JE3BkvgK2yRu.jpg @@ -0,0 +1 @@ +fake-jpg \ No newline at end of file diff --git a/storage/media-library/temp/LZvUnQohKyX7zYlAWEVEV3OmmaBLNv9I/rMvgr63xbMLBuLxbYH0Qqv7sEkM4cYcethumb.jpg b/storage/media-library/temp/LZvUnQohKyX7zYlAWEVEV3OmmaBLNv9I/rMvgr63xbMLBuLxbYH0Qqv7sEkM4cYcethumb.jpg new file mode 100644 index 0000000..7e754f4 --- /dev/null +++ b/storage/media-library/temp/LZvUnQohKyX7zYlAWEVEV3OmmaBLNv9I/rMvgr63xbMLBuLxbYH0Qqv7sEkM4cYcethumb.jpg @@ -0,0 +1 @@ +fake-jpg \ No newline at end of file diff --git a/storage/media-library/temp/XWReSlym4bRuIGCJxIj2XtlonYLGutoF/Qoy0e3w5KBGEIZbNeHAQCryBgYtKa297thumb.jpg b/storage/media-library/temp/XWReSlym4bRuIGCJxIj2XtlonYLGutoF/Qoy0e3w5KBGEIZbNeHAQCryBgYtKa297thumb.jpg new file mode 100644 index 0000000..7e754f4 --- /dev/null +++ b/storage/media-library/temp/XWReSlym4bRuIGCJxIj2XtlonYLGutoF/Qoy0e3w5KBGEIZbNeHAQCryBgYtKa297thumb.jpg @@ -0,0 +1 @@ +fake-jpg \ No newline at end of file diff --git a/storage/media-library/temp/XWReSlym4bRuIGCJxIj2XtlonYLGutoF/eS3k80b4GLnuhGzxSNtBXqz5fuyM5tbB.jpg b/storage/media-library/temp/XWReSlym4bRuIGCJxIj2XtlonYLGutoF/eS3k80b4GLnuhGzxSNtBXqz5fuyM5tbB.jpg new file mode 100644 index 0000000..7e754f4 --- /dev/null +++ b/storage/media-library/temp/XWReSlym4bRuIGCJxIj2XtlonYLGutoF/eS3k80b4GLnuhGzxSNtBXqz5fuyM5tbB.jpg @@ -0,0 +1 @@ +fake-jpg \ No newline at end of file diff --git a/storage/media-library/temp/bUSUtdJl4kj1X0DdugUygG3vvklf6imo/2IDt9E5tJc5CnD0Q4EWgXCwpNBTwpTlQ.jpg b/storage/media-library/temp/bUSUtdJl4kj1X0DdugUygG3vvklf6imo/2IDt9E5tJc5CnD0Q4EWgXCwpNBTwpTlQ.jpg new file mode 100644 index 0000000..7e754f4 --- /dev/null +++ b/storage/media-library/temp/bUSUtdJl4kj1X0DdugUygG3vvklf6imo/2IDt9E5tJc5CnD0Q4EWgXCwpNBTwpTlQ.jpg @@ -0,0 +1 @@ +fake-jpg \ No newline at end of file diff --git a/storage/media-library/temp/bUSUtdJl4kj1X0DdugUygG3vvklf6imo/qwSWDg90QY1xj5uQNxyvvQBem6WdYjTbthumb.jpg b/storage/media-library/temp/bUSUtdJl4kj1X0DdugUygG3vvklf6imo/qwSWDg90QY1xj5uQNxyvvQBem6WdYjTbthumb.jpg new file mode 100644 index 0000000..7e754f4 --- /dev/null +++ b/storage/media-library/temp/bUSUtdJl4kj1X0DdugUygG3vvklf6imo/qwSWDg90QY1xj5uQNxyvvQBem6WdYjTbthumb.jpg @@ -0,0 +1 @@ +fake-jpg \ No newline at end of file diff --git a/storage/media-library/temp/o1H5qF5gy9bn6mZgA4kyoMehDVv3RZnj/OKF1ndSu73Qgv7d7fsrmwZyMK8x2g9dKthumb.jpg b/storage/media-library/temp/o1H5qF5gy9bn6mZgA4kyoMehDVv3RZnj/OKF1ndSu73Qgv7d7fsrmwZyMK8x2g9dKthumb.jpg new file mode 100644 index 0000000..7e754f4 --- /dev/null +++ b/storage/media-library/temp/o1H5qF5gy9bn6mZgA4kyoMehDVv3RZnj/OKF1ndSu73Qgv7d7fsrmwZyMK8x2g9dKthumb.jpg @@ -0,0 +1 @@ +fake-jpg \ No newline at end of file diff --git a/storage/media-library/temp/o1H5qF5gy9bn6mZgA4kyoMehDVv3RZnj/xMqBY4WYOHw1ZB9qz4GLeTCEqIyO0yyp.jpg b/storage/media-library/temp/o1H5qF5gy9bn6mZgA4kyoMehDVv3RZnj/xMqBY4WYOHw1ZB9qz4GLeTCEqIyO0yyp.jpg new file mode 100644 index 0000000..7e754f4 --- /dev/null +++ b/storage/media-library/temp/o1H5qF5gy9bn6mZgA4kyoMehDVv3RZnj/xMqBY4WYOHw1ZB9qz4GLeTCEqIyO0yyp.jpg @@ -0,0 +1 @@ +fake-jpg \ No newline at end of file diff --git a/tests/Feature/WorkermanQueueTest.php b/tests/Feature/WorkermanQueueTest.php new file mode 100644 index 0000000..84a3e94 --- /dev/null +++ b/tests/Feature/WorkermanQueueTest.php @@ -0,0 +1,74 @@ + 'database']); + + Log::spy(); + + dispatch(new PingJob); + + $this->assertSame(1, DB::table('jobs')->count()); + + $this->runWorkerOnce('database', 'default'); + + $this->assertSame(0, DB::table('jobs')->count()); + Log::shouldHaveReceived('info')->withArgs(fn ($message) => str_contains($message, 'WM_PING_CONSUMED')); + } + + /** + * 多队列（default,ai）支持：ai 队列任务也能被消费。 + */ + public function test_worker_consumes_multiple_queues(): void + { + config(['queue.default' => 'database']); + + Log::spy(); + + dispatch((new PingJob)->onQueue('ai')); + + $this->runWorkerOnce('database', 'default,ai'); + + $this->assertSame(0, DB::table('jobs')->count()); + Log::shouldHaveReceived('info')->withArgs(fn ($message) => str_contains($message, 'WM_PING_CONSUMED')); + } + + private function runWorkerOnce(string $connection, string $queues): void + { + $worker = new QueueWorker( + app('queue'), + app('events'), + app(ExceptionHandler::class), + fn () => app()->isDownForMaintenance(), + ); + + $options = new WorkerOptions( + name: 'laralog-test', + backoff: 0, + memory: 128, + timeout: 30, + sleep: 0, + maxTries: 2, + ); + + $worker->runNextJob($connection, $queues, $options); + } +} diff --git a/tests/Fixtures/PingJob.php b/tests/Fixtures/PingJob.php new file mode 100644 index 0000000..0f1f2a9 --- /dev/null +++ b/tests/Fixtures/PingJob.php @@ -0,0 +1,17 @@ +