消息队列的概念、原理和场景
在高并发的时候，程序往往无法做到及时的处理。我们引入一个中间的系统，来进行分流和减压。
所以从本质上讲：消息队列就是一个队列结构的中间件。也就是说，你把消息和内容放入这个容器之后就可以直接返回，不用等它后期处理的结果。另外会有一个程序，读取这些数据并按照顺序处理。
1、队列结构的中间件
2、消息放入后，不必立即处理
3、由订阅者/消费者按顺序处理
也就是说：当遇到一个比较大或者耗时比较长的环节的时候，而同时你的业务又不需要立即知道这个环节的结果，使用消息队列是好的选择。
知识付费的拼团功能使用的就是消息队列功能；把每个拼团订单都储存在消息队列中，拼团完成或拼团结束就可以自动处理这个订单。
application\index\controller\PushJob
/**
* 一个使用了队列的 action
*/
public static function actionWithDoPinkJob(array $data, string $name = '')
{
try {
// 1.当前任务将由哪个类来负责处理。
// 当轮到该任务时，系统将生成一个该类的实例，并调用其 fire 方法
$jobHandlerClassName = 'app\index\job\PullDoPink';
// 2.当前任务归属的队列名称，如果为新队列，会自动创建
$jobQueueName = Config::get('queue_name', '') ? Config::get('queue_name', '') : 'doPinkJobQueue';
// 3.当前任务所需的业务数据 . 不能为 resource 类型，其他类型最终将转化为json形式的字符串
// ( jobData 为对象时，存储其public属性的键值对 )
if ($name) {
$jobData = ['pinkInfo' => $data, 'time' => date('Y-m-d H:i:s'), 'doName' => $name];
$isPushed = Queue::push($jobHandlerClassName, $jobData, $jobQueueName);
} else {
$jobData = ['pinkInfo' => $data, 'time' => date('Y-m-d H:i:s')];
if (!isset($data['pink_time']) || !$data['pink_time']) return true;
$timewait = $data['pink_time'] + 300;
//$jobData = [ 'pinkInfo' => 'hahah', 'time' => date('Y-m-d H:i:s'), 'b' => 21] ;
//$timewait = 20;
// 4.将该任务推送到消息队列，等待对应的消费者去执行
$isPushed = Queue::later($timewait, $jobHandlerClassName, $jobData, $jobQueueName);
//$isPushed = Queue::push($jobHandlerClassName , $jobData , $jobQueueName );
// database 驱动时，返回值为 1|false ; redis 驱动时，返回值为 随机字符串|false
}
if ($isPushed !== false) {
return 1;
} else {
return 1;
}
} catch (ErrorException $e) {
echo $e->getMessage();
}
}application\index\job\PullDoPink
/**
* fire方法是消息队列默认调用的方法
* @param Job $job 当前的任务对象
* @param array|mixed $data 发布任务时自定义的数据
*/
public function fire(Job $job, $data)
{
// 有些消息在到达消费者时,可能已经不再需要执行了
$isJobStillNeedToBeDone = $this->checkDatabaseToSeeIfJobNeedToBeDone($data);
if (!$isJobStillNeedToBeDone) {
$job->delete();
return;
}
if (isset($data['doName']) && $data['doName']) {
$doName = $data['doName'];
$isJobDone = $this->$doName($data);
} else
$isJobDone = $this->doPinkJob($data);
if ($isJobDone) {
// 如果任务执行成功， 记得删除任务
$job->delete();
//print("<info>Hello Job has been done and deleted"."</info>\n");
} else {
if ($job->attempts() > 3) {
//通过这个方法可以检查这个任务已经重试了几次了
// print("<warn>Hello Job has been retried more than 3 times!"."</warn>\n");
$job->delete();
// 也可以重新发布这个任务
//print("<info>Hello Job will be availabe again after 2s."."</info>\n");
//$job->release(2); //$delay为延迟时间，表示该任务延迟2秒后再执行
}
}
} /**
* 有些消息在到达消费者时,可能已经不再需要执行了
* @param array|mixed $data 发布任务时自定义的数据
* @return boolean 任务执行的结果
*/
private function checkDatabaseToSeeIfJobNeedToBeDone($data)
{
return true;
}拼团使用，在订单生成完成后，把参数加入$do_job_pink数组中
PushJob::actionWithDoPinkJob($do_job_pink);在application\index\job\PullDoPink下加如下面的方法用来接受处理数据
/**
* 根据消息中的数据进行实际的业务处理...
*/
private function doPinkJob($data)
{
$pink_id = $data['pinkInfo']['pink_id'];
if ($pink_id) {
$pink_info = \app\wap\model\store\StorePink::where(['id' => $pink_id, 'k_id' => 0, 'status' => 1])->find();
if ($pink_info ? $pink_info = $pink_info->toArray() : []) {
list($pinkAll, $pinkT, $count, $idAll) = \app\wap\model\store\StorePink::getPinkMemberAndPinkK($pink_info);
\app\wap\model\store\StorePink::PinkFail($pink_info['uid'], $idAll, $pinkAll, $pinkT, $count, 1, [], true, true);
}
}
return true;
}如果需要加新的消息队列可以设置不同的名称即可
PushJob::actionWithDoPinkJob($do_job_pink,’名称’);
application\index\job\PullDoPink：
private function 名称($data)
{
return true;
}
完成后重新启动消息队列
