前段时间接了个需求，用户提交表单之后要发邮件、写日志、还要同步一份数据到第三方接口。一开始我图省事，全塞在接口里同步跑，结果接口响应从 200ms 直接飙到 3 秒多，用户点提交转圈转到怀疑人生。后来改成用 Redis 做消息队列，把耗时操作丢到后台异步处理，接口响应立马回到 200ms 以内。这篇就把这套简单实现整理一下，代码都是能直接跑的，希望能帮到遇到同样问题的朋友。
为什么选 Redis 而不是 RabbitMQ
说实话，如果项目体量大、对消息可靠性要求高，RabbitMQ 或者 Kafka 肯定是更正规的选择。但很多中小项目其实用不上那么重的东西，Redis 本身项目里就在用（缓存、Session 都靠它），再拿来做轻量级队列，几乎零额外成本。
Redis 做队列主要有两种思路：
- List 结构：用 LPUSH 入队、BRPOP 出队，简单直接，适合大多数场景。
- Stream 结构：Redis 5.0 之后引入，支持消费组和 ACK 机制，可靠性更好，但用起来稍复杂。
下面我以 List 方案为主，因为它的代码量最少，理解起来也最直观。
整体思路先捋清楚
整个流程其实就三步：生产者把任务序列化成 JSON 塞进 Redis 列表，消费者用阻塞方式不断取任务，取到之后执行具体业务逻辑。中间如果执行失败，可以重新入队或者记录到失败队列里人工处理。
flowchart LR
A[Web 接口] –>|LPUSH| B[(Redis List)]
B –>|BRPOP| C[消费者进程]
C –> D{执行成功?}
D –>|是| E[记录日志]
D –>|否| F[重试或进失败队列]
环境准备
我本地用的是 PHP 7.4 + Redis 6.2，扩展装的是 phpredis。如果你用的是 Predis 这种纯 PHP 客户端，代码稍微改一下连接部分就行，逻辑是一样的。
# 安装 phpredis 扩展
pecl install redis
确认扩展已加载
php -m | grep redis
生产者：把任务丢进队列
生产者这块很简单，核心就是 lPush。我习惯把任务类型、参数、时间戳一起打包，方便消费者那边做区分和排查问题。
<?php
class QueueProducer
{
private $redis;
private $queueKey = 'app:task:queue';
public function __construct()
{
$this->redis = new Redis();
$this->redis->connect('127.0.0.1', 6379);
$this->redis->auth('your_password'); // 没设密码就注释掉
}
public function push(string $type, array $payload): bool
{
$task = [
'id' => uniqid('task_', true),
'type' => $type,
'payload' => $payload,
'created_at' => time(),
'retry' => 0,
];
$json = json_encode($task, JSON_UNESCAPED_UNICODE);
return $this->redis->lPush($this->queueKey, $json) > 0;
}
}
// 调用示例：用户注册后异步发欢迎邮件
$producer = new QueueProducer();
$producer->push('send_welcome_mail', [
'user_id' => 10086,
'email' => 'test@example.com',
]);
这里有个小细节：json_encode 记得加 JSON_UNESCAPED_UNICODE，不然中文参数会变成一堆 \\uXXXX，排查日志的时候看着头疼。
消费者：常驻进程取任务
消费者一般用 CLI 方式跑，配合 supervisor 做进程守护。核心是 brPop，它是阻塞的，队列空的时候会挂起等待，不会空转浪费 CPU。
<?php
class QueueConsumer
{
private $redis;
private $queueKey = 'app:task:queue';
private $maxRetry = 3;
public function __construct()
{
$this->redis = new Redis();
$this->redis->connect('127.0.0.1', 6379);
$this->redis->auth('your_password');
// 阻塞超时设为 5 秒，方便进程优雅退出
$this->redis->setOption(Redis::OPT_READ_TIMEOUT, 5);
}
public function run(): void
{
echo "消费者启动，等待任务…\\n";
while (true) {
// 第二个参数是超时秒数，返回 [key, value] 或空数组
$result = $this->redis->brPop([$this->queueKey], 5);
if (empty($result)) {
continue; // 超时没任务，继续下一轮
}
$task = json_decode($result[1], true);
if (!$task) {
echo "任务格式错误，丢弃\\n";
continue;
}
$this->handle($task);
}
}
private function handle(array $task): void
{
try {
switch ($task['type']) {
case 'send_welcome_mail':
$this->sendMail($task['payload']);
break;
default:
echo "未知任务类型: {$task['type']}\\n";
}
echo "任务 {$task['id']} 执行成功\\n";
} catch (Throwable $e) {
$this->retry($task, $e);
}
}
private function sendMail(array $payload): void
{
// 这里换成你项目里的邮件发送逻辑
sleep(1); // 模拟耗时操作
echo "已发送邮件给 {$payload['email']}\\n";
}
private function retry(array $task, Throwable $e): void
{
$task['retry']++;
echo "任务 {$task['id']} 失败: {$e->getMessage()}\\n";
if ($task['retry'] < $this->maxRetry) {
// 重新入队，稍后重试
$this->redis->lPush($this->queueKey, json_encode($task, JSON_UNESCAPED_UNICODE));
} else {
// 超过重试次数，进失败队列
$this->redis->lPush('app:task:failed', json_encode($task, JSON_UNESCAPED_UNICODE));
echo "任务 {$task['id']} 已进入失败队列\\n";
}
}
}
(new QueueConsumer())->run();
几个容易踩的坑
代码看着简单，但实际跑起来还是有几个地方要注意：
写在最后
这套方案不算什么高深技术，但胜在简单够用，几十行代码就能把接口响应时间降下来。当然它也有局限，比如没有 ACK 机制，消费者取走任务后如果进程突然挂掉，这个任务就丢了。如果业务对可靠性要求高，建议直接上 Redis Stream 或者专业消息队列。
对于大部分中小项目来说，先把同步改异步、把响应速度提上去，比一上来就追求完美架构更实际。等业务量真的上来了，再平滑迁移也不迟。
