我们在开发或者设计一个网站的时候，经常会遇到要短信群发，或者群发email，或者给系统的所有用户发送站内信，或者在订单系统里，我们要记录大量的日志。如果我们的系统是电商系统，在做抢购，秒杀的活动的设计的时候，服务器在高并发下，根本就无法承受这种瞬间的压力等等，很多例子。。。那如果遇到这些问题，如何保证系统能够正常有效的运行，我们该如何去设计，如何去处理呢？
这个时候我们就要用到消息队列来处理这类问题。可以说消息队列是一个中间件，用这种中间件来分流与解压各种并发带来的压力。那么什么是消息队列呢？
耳熟能详的消息队列（原理）
消息队列其实就是一个队列结构的中间件，也就是说把消息和内容放入到一个容器后，就可以直接的返回了，不理会等它后期处理的结果，容器里的内容会有另一个程序按照顺序进行逐个的去处理。
一个消息队列结果是这样的过程：
由一个业务系统进行入队，把消息（内容）逐个插入消息队列中，插入成功之后直接返回成功的结果，然后后续有一个消息处理系统，这个系统会把消息队列中的记录逐个进行取出并且进行处理，进行出队的操作。
消息队列有哪些应用场景
消息队列主要运用在冗余，解耦，流量削峰，异步通讯，还有一些扩展性，排序保证等，下面我们详细来了解一下这些特性
数据冗余
比如一个订单系统，订单很多的时候，到后续需要严格的转换和记录，这个时候消息队列就可以把这些数据持久化存储在队列中，然后由订单处理程序进行获取，后续处理完成之后再把这条记录删除，保证每条记录都能处理完成。
系统解耦
消息队列分离了两套系统：入队系统和出队系统，解决了两套系统深度耦合的问题。使用消息队列后，入队的系统和出队的系统是没有直接的关系的，入队系统和出队系统其中一套系统崩溃的时候，都不会影响到另一个系统的正常运转。
我们用一个系统解耦的案例来详细讲解一下：队列处理订单系统和配送系统
场景：在网购的时候提交订单之后，看到自己的订单货物在配送中，这样就参与进来一个系统是配送系统，如果我们在做架构的时候，把订单系统和配送系统设计到一起，就会出现问题。首先对于订单系统来说，订单系统处理压力较大，对于配送系统来说没必要对这些压力做及时反映，我们没必要在订单系统出现问题的情况下，同时配送系统出现问题，这时候就会同时影响两个系统的运转，所以我们可以用解耦来解决。
这两个系统分开之后，我们可以通过一个队列表来实现两个系统的沟通。首先，订单系统会接收用户的订单，进行订单的处理，会把这些订单写到队列表中，这个队列表是沟通两个系统的关键，由配送系统中的定时执行的程序来读取队列表进行处理，配送系统处理之后，会把已经处理的记录进行标记，这就是整个详细流程。
具体细节设计如下（Mysql队列举例）：
首先，我们用order.php的文件接收用户的订单。
然后生成订单号并对订单进行处理，订单系统处理完成之后会把配送系统需要的数据增加到队列表中。
订单表
| CREATE TABLE `order_queue` ( | |
| `id` int(11) unsigned NOT NULL AUTO_INCREMENT COMMENT '订单的id号', | |
| `order_id` int(11) NOT NULL, | |
| `mobile` varchar(20) NOT NULL COMMENT '用户的手机号', | |
| `address` varchar(100) NOT NULL COMMENT '用户的地址', | |
| `created_at` datetime NOT NULL DEFAULT '0000-00-00 00:00:00' COMMENT '订单创建的时间', | |
| `updated_at` datetime NOT NULL DEFAULT '0000-00-00 00:00:00' COMMENT '物流系统处理完成的时间', | |
| `status` tinyint(2) NOT NULL COMMENT '当前状态，0 未处理，1 已处理，2处理中', | |
| PRIMARY KEY (`id`) | |
| ) ENGINE=InnoDB DEFAULT CHARSET=utf8; |
然后有一个定时脚本，每分钟启动配送处理程序，配送处理程序:goods.php用来处理队列表中的数据，当处理完成之后，会把队列表中的字段状态改为处理完成，这样就结束了整个流程。
具体代码如下:
1 处理订单的order.php文件
| include 'class/db.php'; | |
| if(!empty($_GET['mobile'])){ | |
| $order_id = rand(10000,99999).date("YmdHis").'688'; | |
| $insert_data = array( | |
| 'order_id'=>$order_id, | |
| 'mobile'=>$_GET['mobile'], //记得过滤 | |
| 'created_at'=>date('Y-m-d H:i:s',time()), | |
| 'order_id'=>$order_id, | |
| 'status'=>0, //0，未处理状态 | |
| ); | |
| $db = DB::getIntance(); | |
| //把数据放入队列表中 | |
| $res = $db->insert('order_queue',$insert_data); | |
| if($res){ | |
| echo $insert_data['order_id']."保存成功"; | |
| }else{ | |
| echo "保存失败"; | |
| } | |
| }else{ | |
| echo "1"; | |
| } | |
配送系统处理订单的文件goods.php
| //配送系统处理订单并进行标记 | |
| include 'class/db.php'; | |
| $db = DB::getIntance(); | |
| //1：先要把要处理的数据状态改为待处理 | |
| $waiting = array('status'=>0,); | |
| $lock = array('status'=>2,); | |
| $res_lock = $db->update('order_queue',$lock,$waiting,2); | |
| //2：选择出刚刚更新的数据，然后进行配送系统处理 | |
| if($res_lock){ | |
| //选择出要处理的订单内容 | |
| $res = $db->selectAll('order_queue',$lock); | |
| //然后由配货系统进行处理.....等操作 | |
| //3：把处理过的改为已处理状态 | |
| $success = array( | |
| 'status'=>1, | |
| 'updated_at'=>date('Y-m-d H;i:s',time()), | |
| ); | |
| $res_last = $db->update('order_queue',$success,$lock); | |
| if($res_last){ | |
| echo "处理成功:".$res_last; | |
| }else{ | |
| echo "处理失败:".$res_last; | |
| } | |
| }else{ | |
| echo "全部处理完成"; | |
| } | |
定时执行脚本的goods.sh，每分钟执行一次
| date "+%G-%m-%d %H:%M:%S" //当前年月日 | |
| cd /data/wwwroot/default/mq/ | |
| php goods.php |
然后crontab任务定时执行脚本，并创建日志文件，然后指定输出格式
*/1 * * * * /data/wwwroot/default/mq/good.sh >> /data/wwwroot/default/mq/log.log 2>&1 //指定脚本目录并格式化输出//当然要创建log.log文件
监控日志
tail -f log.log //监控日志
这样订单系统和配送系统j就是相互独立的咯，并不影响另一个系统的正常运行，这就是系统解耦处理.
流量削峰
这种场景最经典的就是秒杀和抢购，这种情况会出现很大的流量剧增，大量的需求集中在短短的几秒内，对服务器的瞬间压力非常大，我们配合缓存redis使用消息队列来有效的解决这种瞬间访问量，防止服务器顶不住而崩溃。
我们也用一个案例来了解了解：使用Redis的List类型实现秒杀。
我们会用到redis的这些函数：
RPUSH/RPUSHX：将值插入到链表的尾部。同上，位置相反
LPOP：移除并获取链表中的第一个元素。
RPOP：移除并获取链表中最后一个元素。
LTRIM：保留指定区间内的元素。
LLEN：获取链表的长度。
LSET:用索引设置链表元素的值。
LINDEX：通过索引获取链表中的元素。
LRANGE：获取链表指定范围内的元素
场景
记录哪个用户参与了秒杀，同时记录时间，这样方便后续处理，用户的ID会存储到【Redis】的链表里进行排队，比如打算让前10个人秒杀成功，后面的人秒杀失败，这样让redis链表的长度保持为10就可以了，10个以后如果再到redis请求追加数据，那么程序上拒绝请求，在redis存取之后，后面的程序会对redis进行取值，因为数据不能长久放在缓存，后面有一个程序遍历处理redis的值，放入数据库永久保存，因为秒杀本来不会太长，可以用脚本循环扫描。
详细说明：
首先Redis程序会把用户的请求数据放入redis，主要是uid和微秒时间戳；然后检查redis链表的长度，超出长度就放弃处理；死循环数据读取redis链表的内容，入库。
秒杀记录表设计：
| CREATE TABLE `redis_queue` ( | |
| `id` int(10) unsigned NOT NULL AUTO_INCREMENT, | |
| `uid` int(11) NOT NULL DEFAULT '0', | |
| `time_stamp` varchar(24) NOT NULL, | |
| PRIMARY KEY (`id`) | |
| ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8; |
接收用户请求的程序：
| $redis = new Redis(); | |
| $redis->connect('127.0.0.1',6379); | |
| $redis-_name = 'miaosha'; | |
| //秒杀用户涌入模拟,500个用户 | |
| for ($i =0; $i < 500; $i++) { | |
| $uid = rand(1000000,99999999); | |
| } | |
| //检查redis链表长度（已存在数量) | |
| $num = 10; | |
| if ($redis->lLen($redis_name) < 10 ) { | |
| //加入链表尾部 | |
| $redis->rPush($redis_name, $uid.'%'.microtime()); | |
| } else { //如果达到10个 | |
| //秒杀结束 | |
| } | |
| $redis->close(); |
处理程序（拿到redis数据写入数据表里）
| //从队列头部读一个值，判断这个值是否存在，如果存在则切割出时间、uid保存到数据库中。（对于redis而言，如果从redis取出这个值，那么这个值就不在redis队列里了，如果出现问题失败了，那么我们需要有一个机制把失败的数据重新放入redis链表中） | |
| $redis = new Redis(); | |
| $redis->connect('127.0.0.1',6379); | |
| $redis-_name = 'miaosha'; | |
| //死循环检测redis队列 | |
| while(1) { | |
| $user = $redis->lpop($redis_name); | |
| if (!$user || $user == 'null') { //如果没有数据跳出循环 | |
| //如果一直执行，速度是非常快的，那么服务器压力大，这里2秒一次 | |
| sleep(2); | |
| //跳出循环 | |
| continue; | |
| } | |
| //拿出微秒时间戳和uid | |
| $user_arr = explode('%', $user); | |
| $insert_data = array( | |
| 'uid' => $user_arr[0]; | |
| 'time_stamp' => $user_arr[1]; | |
| ); | |
| $res = $db->insert('redis_queue', $insert_data); | |
| //如果插入失败 | |
| if (!$res) { | |
| //从哪个方向取出，从哪个方向插回 | |
| $redis->lpush($redis_name, $user); | |
| sleep(2); | |
| } | |
| } | |
| $redis->close(); |
测试的话，可以先执行循环检测脚本，然后再执行秒杀脚本开始测试，监测Mysql数据库的变化。
异步通讯
消息本身可以使入队的系统直接返回，所以实现了程序的异步操作，因此只要适合于异步的场景都可以使用消息队列来实现。
下面来看具体案例：
基本知识点
重点用到了以下命令实现我们的消息推送。
- brpop 阻塞模式 从队列右边获取值之后删除
- brpoplpush 从队列A的右边取值之后删除，从左侧放置到队列B中
逻辑分析
- 在普通的任务脚本中写入push_queue队列要发送消息的目标，并为目标设置一个要推送的内容，永不过期
- RedisPushQueue中brpoplpush处理，处理后的值放到temp_queue，主要防止程序崩溃造成推送失败
- RedisAutoDeleteTempqueueItems处理temp_queue,这里用到了brpop
代码： 普通任务脚本
| foreach ($user_list as $item) { | |
| //命名规则 业务类型_操作_ID_随机6位 值 自定义 我自定义的是"推送内容" | |
| $k_name = 'rabbit_push_' . $item['uid'].'_'.rand(100000,999999); | |
| $redis->lPush('push_queue',$k_name);//左进队列 | |
| $redis->set($k_name, '推送内容'); | |
| } |
RedisPushQueue
| //消息队列处理推送~ | |
| // | |
| // 守护进程运行 | |
| // nohup php YOURPATH/RedisPushQueue.php & 开启守护进程运行，修改文件之后需要从新启动 | |
| // blpop 有值则回去 没值则阻塞 主要就是这个函数在起作用 不过并不安全，程序在执行过程中崩溃就会导致队列中的内容 | |
| // 永久丢失~ | |
| // BRPOPLPUSH 阻塞模式 右边出 左边进 在填写队列内容的时候要求从左进入 | |
| // | |
| ini_set('default_socket_timeout', -1); //不超时 | |
| require_once 'YOURPARH/Rongcloud.php'; | |
| $redis = new Redis(); | |
| $redis->connect('127.0.0.1', 6379); | |
| $redis->select(2);//切换到db2 | |
| $redis->setOption(Redis::OPT_READ_TIMEOUT, -1); | |
| // temp_queue临时队列防止程序崩溃导致队列中内容丢失 0代表永不超时！ | |
| While ($key = $redis->brpoplpush('push_queue', 'temp_queue', 0)) { | |
| if ($val = $redis->get($key)) { | |
| //rabbit_push_20_175990 | |
| $arr = explode('_', $key); | |
| if (count($arr) != 4) { | |
| continue; | |
| } | |
| $id = $arr[2]; | |
| push($id, $val); | |
| //删除key内容 | |
| $redis->del($key); | |
| } | |
| } | |
| function push($id, $v) | |
| { | |
| //推送操作~ | |
| } |
RedisAutoDeleteTempqueueItems
自动处理temp_queue中的元素，这个操作是防止RedisPushQueue崩溃的时候做处理。
处理思路是 使用brpop 命令阻塞处理temp_queue这个队列中的值，如果能获取到”值”对应的”值”，说明RedisPushQueue执行失败了，将值还lpush到push_queue中，以备从新处理
至于为什么使用brpop命令，是因为在RedisPushQueue中我们使用的是brpoplpushnohup php YOURPATH/RedisAutoDeleteTempqueueItems.php & 开启守护进程运行，修改文件之后需要从新启动
| ini_set('default_socket_timeout', -1); //不超时 | |
| $redis = new Redis(); | |
| $redis->connect('127.0.0.1', 6379); | |
| $redis->select(2);//切换到db2 | |
| $redis->setOption(Redis::OPT_READ_TIMEOUT, -1); | |
| while($key_arr = $redis->brPop('temp_queue',0)){ | |
| if(count($key_arr) != 2){ | |
| continue; | |
| } | |
| $key =$key_arr[1]; | |
| if($redis->get($key)){//能获取到值 说明RedisPushQueue执行失败 | |
| $redis->lPush('push_queue',$key); | |
| } | |
| } |
更专业的消息队列，你可以使用：RabbitMQ，ctiveMq，eroMq，Kafka，这里就不过多的去介绍这些了。