当前位置:首页 > PHP专题 > 正文

耳熟能详的消息队列你如何用,应用在哪些场景?

04-07 PHP专题 博客


我们在开发或者设计一个网站的时候,经常会遇到要短信群发,或者群发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文件 
<?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
<?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,每分钟执行一次
#!/bin/bash  
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;

接收用户请求的程序:
<?php

$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数据写入数据表里)
<?php
//从队列头部读一个值,判断这个值是否存在,如果存在则切割出时间、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

代码: 普通任务脚本
<?php
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
<?php
//消息队列处理推送~
//
 // 守护进程运行
 // 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 & 开启守护进程运行,修改文件之后需要从新启动
<?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,ActiveMq,ZeroMq,Kafka,这里就不过多的去介绍这些了。
 

以上是本文的全部内容,希望对大家的学习有帮助,也希望大家多多支持 php自学中心 感谢阅读!