必须显式配置 async-queue 的 Redis 连接池、创建无副作用的可重试 Job 类、启用并调优 Consumer 进程，并通过 DriverFactory 投递任务；否则易导致消息丢失、连接暴增或静默失败。
要在 Hyperf 3.1 中稳定运行异步队列并确保日志、通知等后台任务不丢失、不阻塞主流程，必须让 async-queue 与 Redis 的连接、序列化、超时、并发模型完全对齐，稍有错配就会出现消息堆积、重复消费或静默丢弃。
确认 Redis 连接池已就绪且被 async-queue 正确复用
async-queue 默认不自动复用 hyperf/redis 配置的连接池，必须显式声明，否则会新建独立连接，导致连接数暴增、超时不可控。打开 config/autoload/async_queue.php，在 default 驱动配置中加入 redis 子项：
'redis' => [
'pool' => 'default',
],
这一步不能省略——Hyperf 3.1+ 的 async-queue 不再默认继承全局 redis 配置，【漏写 'redis' => ['pool' => 'default'] 将导致队列进程使用未配置的直连模式，极大概率触发 read_timeout 中断】。若你项目中 Redis 使用了非 default 别名（如 cache 或 queue），此处必须严格对应。
创建可重试、可诊断的 Job 类
Job 是 async-queue 的执行单元，它不是普通类，其构造函数参数会被序列化存入 Redis，handle() 方法在独立 Consumer 进程中反序列化后执行。因此必须规避资源句柄和闭包。
方法一：标准 Job 继承方式
执行命令生成骨架：
php bin/hyperf.php gen:job LogWriteJob
编辑 app/Job/LogWriteJob.php，关键修改如下：
public function __construct(public array $logData, public string $operatorId)
{
// 禁止在此处查数据库、发 HTTP 请求、打开文件
// $this->logData 必须是纯数组，不含对象或资源
}
public function handle()
{
// 所有 IO 操作必须放在这里，且需 try-catch 包裹
try {
$pdo = ApplicationContext::getContainer()->get(PDO::class);
$pdo->prepare('INSERT INTO sys_logs (...) VALUES (...)')->execute([...]);
} catch (\Throwable $e) {
// 记录到本地日志，避免吞掉错误
logger()->error('LogWriteJob failed', ['error' => $e->getMessage(), 'data' => $this->logData]);
throw $e;
}
}
启用 Consumer 进程并调优并发粒度
Consumer 进程是 async-queue 的心脏，它持续从 Redis pop 任务并执行。Hyperf 3.1 默认不启用该进程，必须手动注册。
第一步：确认 config/autoload/processes.php 中已包含：
Hyperf\AsyncQueue\Process\ConsumerProcess::class,
第二步：调整并发参数防止 DB 连接打满。打开 config/autoload/async_queue.php，修改 default 配置：
'concurrent' => [
'limit' => 3,
],
'processes' => 2,
这里设为 3 并发 + 2 进程，等效于最多 6 个任务并行执行。若你的 MySQL 连接池最大仅 10，这个值就是安全上限；设高了会导致 PDOException: Too many connections。不要迷信“越多越快”，实测超过 5 并发后单任务平均耗时反而上升。
第三步：启动服务并验证进程存活：
php bin/hyperf.php start
ps aux | grep ConsumerProcess
投递任务并验证是否进队
任务必须通过容器内获取的 Queue 对象投递，不能直接 new Job 后调用 handle —— 那样等于同步执行，完全绕过队列。
在控制器或 Service 中注入 Queue：
use Hyperf\AsyncQueue\Driver\DriverFactory;
use Hyperf\AsyncQueue\JobInterface;
投递代码示例：
$job = new LogWriteJob([
'uri' => '/admin/user/edit',
'method' => 'PUT',
'ip' => $request->getServerParams()['remote_addr'] ?? '',
], 'U10086');
$driver = $this->container->get(DriverFactory::class)->get('default');
$driver->push($job);
验证是否成功入队：用 redis-cli 直连后执行
lrange plan_queue 0 -1
（plan_queue 是你在 async_queue.php 中配置的 channel 值）—— 若返回空数组，说明 push 失败或被静默拦截；若返回一串序列化字符串，则任务已在队列中等待消费。
