必须显式配置 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 失败或被静默拦截;若返回一串序列化字符串,则任务已在队列中等待消费。


















