详解PHP消息队列的实现以及运用(附流程图)
消息队列的概念、原理、实现方式
概念
- 队列结构的一个中间件
- 不需要立即消费消息
- 由消费者或者订阅者进行按顺序消费
基本的流程图如下所示
- 流程
应用场景
- 冗余
- 解耦
- 流量削峰
- 异步通信
实现方式
- mysql:可靠、速度慢
- redis:速度快,对于大消息包处理较慢
- 消息系统:可靠、专业性强
消息的触发机制
- 死循环的方式,故障时无法及时恢复
- 定时任务:压力均分、但是处理量有上限
- 守护进程的方式
解耦 (订单和配送系统)
架构设计1 采用定时任务的方式
php入门到就业线上直播课:进入学习
Apipost = Postman + Swagger + Mock + Jmeter 超好用的API调试工具:点击使用使用配送处理系统进行处理时,将当前数据库里需要处理的订单状态更新为2,待处理完成后将状态设为1
可以每次指定更新多少条数据
流量削锋 (redis实现秒杀)
使用队列的数据结构
- lpush/rpush 将数据放入列表中
- lpop/rpop 将数据移除列表并获取到移除的值
- ltrim 保留指定区间内的元素
- llen 获取列表长度
- lset 通过索引设置列表的值
- lindex 通过索引获取列表中的值
- lrange 获取指定范围的元素
图示如下
代码流程如下
秒杀程序将请求写入redis(uid,time)
检查redis列表存放的长度,超过10个直接舍弃
通过死循环读取redis数据,并存入数据库
// Spike.php 秒杀程序if(Redis::llen('lottery') < 10){ // 成功 Redis::lpush('lottery', $uid.'%'.microtime());}else{ // 失败}
登录后复制// Warehousing.php 入库程序while(true){ $user = Redis::rpop('lottery'); if (!$user || $user == 'nil') { sleep(2); continue; } $user_arr = explode($user, '%'); $insert_user = [ 'uid' => $user_arr[0], 'time' => $user_arr[1] ]; $res = DB::table('lottery_queue')->insert($insert_user); if (!$res) { Redis::lpush('lottery', $user); }}
登录后复制上述代码中假如并发过大的话会存在超卖的情况,此时可以使用文件锁或者redis分布式锁进行控制,先将商品放入redis list中 使用rpop进行取出,如果取不到则说明已经卖完
具体的思路及伪代码如下
// 先将商品放入redis中 $goods_id = 2; $sql = select id,num from goods where id = $goods_id; $res = DB::select($sql); if (!empty($res)) { // 也可以指定多少件 Redis::del('lottery_goods' . $goods_id); for($i=0;$i<$res['num'];$i++){ Redis::lpush('lottery_goods . $goods_id', $i); } LOG::info('商品存入队列成功,数量:' . Redis::llen('lottery_goods . $goods_id')); } else { LOG::info($goods_id . '加入失败'); }
登录后复制// 开始秒杀 $count = Redis::rpop('lottery_goods' . $goods_id); if (!$count) { // 商品已抢完 ... } // 用户抢购队列 $user_list = 'user_goods_id_' . $goods_id; $user_status = Redis::sismember($user_list, $user_id); if ($user_status) { // 已抢过 ... } // 将抢到的放到列表中 Redis::sadd($user_list, $uid); $msg = '用户:' . $uid . '顺序' . $count; Log::info($msg); // 生成订单等 ... // 减库存 $sql = update goods set num = num -1 where id = $goods_id and num > 0; // 防止超卖 DB::update($sql) // 抢购成功
登录后复制rabbitmq
架构及原理
其中P代表生产者,X为交换机(channal),C代表消费者简单使用
// Send.php require_once __DIR__.'/vendor/autoload.php'; use PhpAmqpLib\Connection\AMQPStreamConnection; use PhpAmqpLib\Message\AMQPMessage; $connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest'); // 创建通道 $channel = $connection->channel(); // 声明一个队列 $channel->queue_declare('user_email', false, false, false, false); // 制作消息 $msg = new AMQPMessage('send email'); // 将消息推送到队列 $channel->basic_publish($msg, '', 'user_email'); echo '[x] send email'; $channel->close(); $connection->close();
登录后复制// Receive.php require_once __DIR__.'/vendor/autoload.php'; use PhpAmqpLib\Connection\AMQPStreamConnection; use PhpAmqpLib\Message\AMQPMessage; $connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest'); //创建通道 $channel = $connection->channel(); $channel->queue_declare('user_email', false, false, false, false); // 当收到消息时的回调函数 $callback = function($msg){ //发送邮件 echo 'Received '.$msg->body.'\n'; }; $channel->basic_consume('user_email', '', false, true, false, false, $callback); // 保持监听状态 while($channel->is_open()){ $channel->wait(); }
登录后复制相关阅读 更多-
WebStorm干嘛用的 WebStorm和VSCode哪个好用 时间:2025-09-13
-
PyCharm详细的安装及使用教程 时间:2025-09-13
-
PyCharm是干什么用的 PyCharm和Python的区别 时间:2025-09-13
-
PHP运行环境的搭建方法及流程详解 时间:2025-09-13
-
PHPstorm环境配置与应用 PHPstorm怎么配置PHP环境 时间:2025-09-13
-
PHP date()函数详解(定义、语法、用法) 时间:2025-09-13
-
今日更新
-
想知道天空的梗是什么梗吗?揭秘爆火网络热梗背后的有趣故事!
阅读:18
-
天空梗是什么梗?揭秘网络爆火天空表情包的搞笑起源与玩法
阅读:18
-
想知道天籁是什么梗?揭秘网络热词天籁的爆笑来源和流行用法!
阅读:18
-
想知道天蓝怪是什么梗吗?揭秘这个火爆网络的神秘热词,让你秒懂潮流!
阅读:18
-
天蓝怪行为是什么梗?揭秘网络爆火神秘现象背后的搞笑真相!
阅读:18
-
重返未来:1999新角色野树莓-野树莓抽取建议
阅读:18
-
以闪亮之名全新主线之旅-章节第36章即将开启
阅读:18
-
梦幻西游恶魔泡泡怎么获得-恶魔泡泡获取方法
阅读:18
-
崩坏星穹铁道3.6版本新内容公布-可免费获五星角色
阅读:18
-
天雷滚滚是什么梗?揭秘网络热词背后的爆笑名场面
阅读:18