消息队列 MySQL实现消息队列

Posted 2021-04-26 爱思考的coder

消息队列（一）

mysql实现消息队列

（原创内容，转载请注明来源，谢谢）

一、概述

         消息队列（MessageQueue，通常简称MQ）是一种进程间通信或同一进程的不同线程间的通信方式，是分布式应用间交换信息的一种技术。通过消息队列，应用程序可独立地执行，它们不需要知道彼此的位置、或在继续执行前不需要等待接收程序接收此消息。

         消息队列有多种实现方式，可以用关系型数据库（如Mysql）、Nosql（如redis）、现有框架（如rabbitMQ）等。

Mysql处理消息队列的场景：主要是在数据处理量大、耗时久、处理流程繁杂、处理内容多、需要持久化（入库）、业务处理要求相对不实时的场景，如发邮件、发短信、订单后续处理、操作数据记录日志等。

 

二、场景分析

         现假设有一个订单处理系统，包括用户支付产生订单、货物分配及发货两个子流程。现由于订单产生量大，故如果将产生订单、货物分配合并在一起执行，对于用户而言，将要等待较久的时间才能看到支付成功的页面，用户体验不佳。因此可以将这两个流程分开。

         1）用户支付产生订单

         在校验有货后，将货物表相应的内容取出到临时表，取出成功即返回用户支付成功待发货。

         2）定时轮询临时表

         编写脚本，定时（如10分钟）轮询此临时表，每次取若干条支付成功的数据（如1000条，具体看处理能力及数据量而定，保证处理时间在10分钟内），进行后续处理。

         3）将第二步的数据进行后续处理，如通知物流取货、通知仓库出货，成功后将临时表的这几条数据清除（或将状态置成已发货）。

         根据上述场景，对于有10000个用户同时购买时，也只需要校验是否有货，确认有货就可以给用户返回支付成功页面。而后台的处理即使100分钟（上述第二步假设每10分钟处理1000条），对于用户而言也只是100分钟后看到已发货的状态，相比于在支付页面多花1分钟甚至更多的时间来说，这100分钟反而并算不了什么。

         因此，此场景就非常适合于用Mysql解决此消息队列。

 

三、程序实现

         1）引入数据库处理文件DbDealer.php，此文件在PDO的文章中有详细的描述实现过程，主要是用PDO实现增删改查。

         2）创建MQ工厂类MessageQueueDealer，通过传入的类型判断是哪种方式的MQ（本例使用的是Mysql，但预留了Redis等场景以便后续扩展），以及判断是哪种业务场景（对于本例而言是订单处理场景）。文件名messagequeue.php

<?php

require_once('service/assembly.php');

 

class MessageQueueDealer{

         private static $ins;

         private function__construct(){}

         private function__clone(){}

         public static functiongetInstance(){

                   if(null ==self::$ins || !($ins instanceof MessageQueueDealer)){

                            self::$ins= new MessageQueueDealer();

                   }

                   returnself::$ins;

         }

        

         //mq处理函数type=mysql,redis,..（mq类型）

         public functionmqDealer($type, $condition){

                   switch($type){

                            case'mysql':

                                     $this->mysqlMq($condition);

                                     break;

                            default:

                                     break;

                   }

         }

        

         //mysql的mqcondition=order,...（业务场景）

         private functionmysqlMq($condition){

                   switch($condition){

                            case'order':

                                     $order= new Order();

                                     $orderInfo= $order->transOrder();

                                    

                   }

         }

}       

$mq = MessageQueueDealer::getInstance()->mqDealer('mysql','order');

         3）引入服务中间文件，其用于引用各个处理类，本例引入订单处理类。文件名assembly.php

         <?php

require_once('order.php');

         4）订单处理类Order，用于定时接处理任务。文件名order.php

         <?php

require_once('dbDealer.php');

//订单处理类，将支付成功的订单移入另一个表，并且从本表删除

class Order{

         publicfunction transOrder(){

                   $db= DbDealer::getInstance()->getConnection();

                   $res= $db->setLimit(2)->setOrderBy(array('asc', 'id'))

                                       ->setSelectColumn(' id,name ')->get_mytable_by_status(3);//每次处理两条，把status=3的记录挪走

                   $arrOrder= $res->fetchAll(PDO::FETCH_ASSOC);//内容全部取出成关联数组

                   //两个步骤一起，插入新表以及移除旧表

                   if(!empty($arrOrder)){

                            $maxId= 0;

                            foreach($arrOrderas $order){

                                     $maxId= $maxId < $order['id'] ? $order['id'] : $maxId;

                            }

                            $db->startTrans();

                            $db->insertBatch('mynewtable',$arrOrder);

                            $db->delete('mytable',array('id' => '<'.$maxId));

                            $db->commitTrans();

                   }                

         }

}

         5）编写crontab脚本，让其定时执行messagequeue，以完成订单转移。

         在linux界面输入crontab –e，进入vim的编辑界面，输入：

*/10 * * * * php/usr/share/nginx/html/message_queue/messagequeue.php

 

四、总结

         Mysql实现消息队列的方式较为简单，其在处理非实时的数据时具有较好优势，因为其存取方便，而非实时情况下也不会有大量的数据库连接，防止正常业务因为大量的连接而让数据库服务器奔溃。

——written by linhxx 2017.07.26