refactor: Workerman 消费改用 Laravel Queue Worker,database/redis 统一支持(实测 redis 消费通过);默认消费 default,ai 队列;WORKERMAN_QUEUES/MAX_TRIES/TIMEOUT 配置;2 个队列消费测试

This commit is contained in:
ak
2026-08-11 19:27:43 +08:00
parent 34ea046906
commit 1c8a801238
16 changed files with 164 additions and 54 deletions
+32 -51
View File
@@ -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 队列 Workerdatabase / 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);
}
}