手把手教你用PHP+Redis实现消息队列功能：超详细实战教程

这篇文章是专门给PHP开发者准备的“实战说明书”——从Redis消息队列的核心逻辑讲起（比如选List做简单队列还是Stream搞高可靠场景），手把手教你写代码：如何用PHP生成消息、让消费者稳定取消息，甚至连消息丢失、重复消费这些坑都帮你想好了应对办法！不管是要做异步任务解耦、高并发削峰，还是处理延迟任务，跟着文中的步骤敲代码，你就能快速搭起能用在项目里的Redis消息队列，直接解决实际问题。

做PHP开发的你，是不是碰到过这种糟心事儿？秒杀活动一开始，订单请求铺天盖地涌进来，数据库直接被打崩；或者要给1万用户发邮件，循环发送时第三方接口超时，一半邮件没发出去；再或者异步处理图片压缩，任务堆在那儿，用户等了半小时还没看到缩略图？其实这些问题，用Redis做个消息队列就能解决——不用搭复杂的中间件，几行PHP代码就能搞定，我去年帮3个小项目做过，稳得很。

为什么选PHP+Redis做消息队列？先解决你最关心的「值不值得」问题

先跟你掰扯清楚，为啥选Redis不选其他中间件？不是说RabbitMQ、Kafka不好，而是Redis太适合中小项目了——你项目里大概率已经在用Redis做缓存（比如存用户会话、商品库存），不用额外部署新服务，省了运维成本；而且学习成本低，就几个命令，半小时就能上手。

我去年帮朋友的美食电商做秒杀功能，一开始直接把订单请求写数据库，结果每秒500个请求冲过来，数据库连接池瞬间满了，订单全失败，用户骂得客服都不敢接电话。后来我改成用Redis List做队列：把订单数据先以JSON字符串的形式，用lpush塞到order_queue这个列表里，再用消费者进程用brpop阻塞式取消息，慢慢写数据库。改完之后，秒杀当天没崩过，订单成功率从60%涨到了99%——你看，这就是消息队列的「削峰」作用，把瞬间的高并发摊成平稳的流量。

再跟你对比下其他方案：RabbitMQ要装Erlang环境，运维起来麻烦；Kafka适合大数据量的日志收集，但学习成本高，中小项目用它有点「杀鸡用牛刀」。Redis的优势就是轻量、灵活、贴合中小项目的需求——如果你的场景不需要复杂的路由、事务，只是要异步解耦、高并发削峰，选Redis准没错。

手把手敲代码：从0到1实现PHP+Redis消息队列

接下来直接上硬菜——敲代码！先确保你环境没问题：PHP需要装Redis扩展（推荐phpredis，性能好；或者用predis，不用编译，Composer装就行），Redis服务要启动，端口默认6379。

第一步：写「生产者」——往Redis里塞消息

生产者的作用是把要处理的任务「丢」进队列。比如你要处理订单创建后的「减库存+发通知」任务，生产者代码可以这么写：

<?php 
// 
连接Redis（用phpredis扩展）
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
// 
构造消息内容（JSON字符串，方便传输复杂数据）
$message = [
 'order_id' => uniqid('order_'), // 生成唯一订单ID
 'user_id' => 1001,
 'goods_id' => 2003,
 'amount' => 299,
 'create_time' => time()
];
$jsonMessage = json_encode($message);
// 
往队列里塞消息（lpush：从列表头部添加，保证先进先出）
$redis->lpush('order_queue', $jsonMessage);
echo "消息发送成功：" . $jsonMessage . PHP_EOL;
?>

这里用lpush而不是rpush，是因为后面消费者要用brpop从列表尾部取——这样就是「先进先出（FIFO）」，符合队列的逻辑。

第二步：写「消费者」——从Redis里取消息并处理

消费者的作用是「蹲守」队列，拿到消息后处理业务逻辑。关键是要用阻塞式读取（brpop），避免一直轮询浪费CPU资源。代码示例：

<?php 
// 
连接Redis
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
// 
阻塞式读取队列（brpop：从列表尾部取，超时时间0=一直等）
while (true) {
 // 参数说明：队列名、超时时间（秒）
 $result = $redis->brpop('order_queue', 0);
 if ($result) {
 // $result[0]是队列名，$result[1]是消息内容
 $jsonMessage = $result[1];
 $message = json_decode($jsonMessage, true);
 echo "收到消息：" . $jsonMessage . PHP_EOL;
 try {
 // 
处理业务逻辑（比如减库存、发通知）
 handleOrder($message);
 echo "消息处理成功：order_id=" . $message['order_id'] . PHP_EOL;
 } catch (Exception $e) {
 // 
处理失败：把消息丢进「死信队列」，后续重试
 $redis->lpush('dead_letter_queue', $jsonMessage);
 echo "消息处理失败，已存入死信队列：" . $e->getMessage() . PHP_EOL;
 }
 }
}
// 模拟业务处理函数
function handleOrder($message) {
 // 这里写真实逻辑：比如调用库存服务减库存、调用短信接口发通知
 // 故意模拟一个可能失败的场景（比如第三方接口超时）
 if (rand(1, 10) == 5) { // 10%的失败概率
 throw new Exception("第三方物流接口超时");
 }
}
?>

这里有两个关键要点：

第三步：进阶——用Redis Stream实现高可靠消息队列

上面的List结构有个小问题：消费者挂了，消息可能会丢。比如消费者取走消息后，还没处理完就崩溃了，消息已经从List里删掉了，找不回来。这时候可以用Redis 5.0推出的Stream结构——它支持「消息确认（ACK）」和「消费者组」，更可靠。

Stream的核心逻辑是：

直接上代码（以订单支付通知为例）：

<?php 
$redis = new Redis();

$redis->connect('127.0.0.1', 6379);
// XADD：往Stream里写消息，表示让Redis自动生成消息ID（格式：时间戳-序列号）
$messageId = $redis->xadd(
 'payment_stream', // Stream键名
 '', // 消息ID（*=自动生成）
 [
 'order_id' => 'order_12345',
 'user_id' => 1001,
 'status' => 'paid',
 'pay_time' => time()
 ]
);
echo "Stream消息发送成功，ID：" . $messageId . PHP_EOL;
?>

<?php 
$redis = new Redis();

$redis->connect('127.0.0.1', 6379);
// 创建消费者组：组名pay_group，起始ID$（表示从最新消息开始）
$redis->xgroup('CREATE', 'payment_stream', 'pay_group', '$', true);
echo "消费者组创建成功" . PHP_EOL;
?>

<?php 
$redis = new Redis();

$redis->connect('127.0.0.1', 6379);
// 从消费者组里取消息（group：组名，consumer：消费者名，count：每次取1条，block：阻塞10秒）
while (true) {
 $result = $redis->xreadgroup(
 'GROUP', 'pay_group', 'consumer_1', // 组名+消费者名
 'COUNT', 1, // 每次取1条
 'BLOCK', 10000, // 阻塞10秒（10000毫秒）
 'STREAMS', 'payment_stream', '>' // >表示取未被处理的消息
 );
 if ($result) {
 $streamData = $result['payment_stream'];
 foreach ($streamData as $messageId => $message) {
 echo "收到Stream消息，ID：" . $messageId . PHP_EOL;
 echo "消息内容：" . json_encode($message) . PHP_EOL;
 try {
 // 处理业务逻辑（比如给用户发支付成功短信）
 sendPaymentNotice($message);
 // 确认消息（XACK：告诉Redis已经处理完）
 $redis->xack('payment_stream', 'pay_group', $messageId);
 echo "消息确认成功" . PHP_EOL;
 } catch (Exception $e) {
 // 处理失败：不用动，消息会留在pending列表里，后续重试
 echo "消息处理失败：" . $e->getMessage() . PHP_EOL;
 }
 }
 }
}
function sendPaymentNotice($message) {
 // 模拟发送短信（故意设置10%的失败概率）
 if (rand(1, 10) == 5) {
 throw new Exception("短信接口超时");
 }
}
?>

Stream的优势很明显：就算消费者崩溃了，未确认的消息还在pending列表里，其他消费者可以接着处理，不会丢——适合需要高可靠的场景（比如订单支付、金融交易）。

最后：性能优化小技巧

要是你按这些步骤做了，肯定能搭出一个能用在项目里的Redis消息队列。我自己用这些代码帮3个小项目解决了异步问题，没出过大乱子——你要是碰到什么坑，或者改了之后有效果，欢迎留言告诉我，咱们一起聊聊！

PHP+Redis做消息队列，适合哪些场景啊？

PHP+Redis的消息队列特别适合中小项目的常见痛点场景，比如秒杀活动的高并发削峰——像文中提到的电商秒杀，把瞬间涌来的订单请求先存进Redis队列，再用消费者慢慢写数据库，避免直接打崩数据库；还有异步任务解耦，比如用户下单后要发邮件、压缩商品图片，这些不用同步等结果的操作，用队列异步处理能提升主流程的响应速度；另外像延迟任务（比如订单超时未支付自动关闭），也能结合Redis的过期键或者定时任务来实现。总之就是中小项目不用额外部署新中间件，刚好解决这些实际问题，成本低又好用。

要是你项目里已经在用Redis做缓存（比如存用户会话、商品库存），那更不用额外折腾，直接复用Redis就行，省了运维和学习成本。

Redis的List和Stream结构，做消息队列有啥区别？

List结构是Redis最基础的队列方案，简单好上手，就几个命令（lpush/brpop），适合轻量场景——比如你要做个简单的异步发邮件，用List完全够。但它有个小缺点：如果消费者取走消息还没处理完就崩溃了，消息已经从List里删掉，找不回来。

Stream是Redis5.0之后的高可靠方案，解决了List的痛点——它支持消息确认（ACK）和消费者组，消费者取消息后要ACK确认，没确认的消息会留在“pending列表”里，就算消费者崩溃了，后续也能重新取出来处理；而且能分成多个消费者组，适合更复杂的场景（比如多个服务需要处理同一条消息）。要是你需要高可靠的场景（比如金融支付的订单通知），选Stream更稳。

消费者进程崩溃了，消息会不会丢啊？

得看你用的是哪种结构。要是用List的话，消费者用brpop取走消息后，List里就没这条消息了——如果这时候消费者崩溃（比如进程被杀掉），没处理完的消息就丢了，找不回来。

但用Stream的话就不会，因为Stream的消息取走后需要ACK确认，没ACK的消息会留在pending列表里，就算消费者崩溃，其他消费者或者重启后的消费者还能从pending列表里取出来重试，消息不会丢。比如文中提到的支付通知场景，用Stream的话，就算消费者进程崩溃，没确认的消息还在，后续处理完再ACK就行。

用Redis做消息队列，怎么优化性能啊？

首先可以开多消费者进程——如果消息量太大，一个消费者处理不过来，比如发邮件要2小时，开4个消费者进程能缩短到20分钟（像文中提到的博客批量发邮件的例子），可以用PHP的pcntl扩展创建子进程，或者用Supervisor管理进程；然后避免大消息，Redis的字符串最大能存512MB，但别塞太大的消息（比如超过1MB），不然会影响性能，尽量把大数据存数据库，消息里只存ID就行；还有要监控队列长度，用LLEN（List）或者XLEN（Stream）看队列里的消息数，如果突然变长，说明消费者处理不过来，得加进程或者优化业务逻辑（比如优化数据库写入速度）。

消息处理失败了怎么办？

可以用“死信队列”来处理——比如文中提到的，当消息处理失败（像第三方接口超时、数据库连接失败），把失败的消息用lpush塞到专门的死信队列（比如dead_letter_queue）里，后续可以写个脚本定时从死信队列里取消息，重新推回主队列重试，或者人工排查失败原因。比如之前帮朋友做电商的时候，第三方物流接口超时导致消息处理失败，把这些消息丢进死信队列，后来排查是接口限流，调整后重试就成功了，避免了消息丢失和用户投诉。

YUANLEISVIP

收藏 链接