PHP && RabbitMQ防止消息丢失

RabbitMQ防止消息丢失，保证消息传递的可靠性，保证每条消息都正常传递，并最终至少消费一次。

背景：订单支付状态同步，微信、支付宝、银联等第三方平台异步回调之后，进入队列，为其他服务调用提供数据。为了保证支付状态同步业务的可用性，肯定不希望有订单在传递过程中丢失。

问题：什么情况下消息可能丢失呢？

角色：生产者、RabbitMQ服务、消费者  （显然，三大主角都有可能演砸）

华丽的下划线  —— 请开始你的表演

主角一出镜：

              我要发送消息到  RabbitMQ服务，但是在去的路上（网络），太过颠簸（网络抖动），把自己丢了。（有时候我自己都佩服我我自己）

解决：

1、采用事务机制，要么成功，要么失败。但是这样吞吐量会降低，影响性能。一般不建议采用（所以也不再提供伪代码）事务同步等待。

2、采用confirm 模式，ack 消息确认，收到nack 者消息重发，做补偿处理。

在生产者设置开启 confirm 模式之后，你每次写的消息都会分配一个唯一的 id，然后如果写入了 RabbitMQ 中，RabbitMQ 会给你回传一个 ack 消息，告诉你消息接收成功。如果 RabbitMQ 没能处理这个消息，会回调你的一个 nack 接口，告诉你消息接收失败，你需要重新发送。在开启持久化之后，消息先到达交换机、队列、并持久化之后，才会回传ack。异步回调信息确认。可以单条信息回复确认一次，也可以多条信息，回复确认一次。

调用的API：（截取代码，详细代码看下文。不要着急尝试）

            //4.1 选择为 confirm 模式（此模式不可以和事务模式 兼容）

            self::$channel->confirm_select();


          //4.2 设置异步回调消息确认 （生产者 防止信息丢失）

            self::$channel->set_ack_handler(

                function (AMQPMessage $message) {

                    echo "Message acked with content " . $message->body . PHP_EOL;

                    self::apiResponse(self::$rabbit_success_code, 'success', $message->body);

                }

            );

            self::$channel->set_nack_handler(

                function (AMQPMessage $message) {

                    echo "Message received failed，Please try again：" . $message->body . PHP_EOL;

                    self::apiResponse(self::$rabbit_err_code, 'Message received failed，Please try again', $message->body);

                }

            );


            //4.3 阻塞等待消息确认 （生产者 防止信息丢失）

            self::$channel->wait_for_pending_acks();

主角二 MQ 服务：

开始我的表演了，如果告诉我消息要持久化，那么我就记录到磁盘，如果不告诉我，我可默认为你不需要持久化，我要是重启，消息可不恢复。

怎么告诉我需要持久化：（两步走）

    声明交换机、队列的时候，参数 durable = true ，让元数据保存

    发送消息的时候，设置 deliveryMode = 2

    第一步、

            //1、声明 交换机

            self::$channel->exchange_declare(self::$cache_exchange, self::EXCHANGE_MODEL, false, true, false);


            //2、声明队列

            self::$channel->queue_declare(self::$cache_queue, false, true, false, false, false);



    第二步、

      $msg = new AMQPMessage($data, array(

                //参数 发送消息的时候将消息的 deliveryMode 设置为 2 （RabbitMQ 持久化 步骤二：消息）

                'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT

            ));

            self::$channel->basic_publish($msg, self::$cache_exchange, self::$cache_queue);

设置之后，我一定保证我接到的数据，肯定会持久化到磁盘，你可以完全放心，我可是著名表演艺术家，从未失误。

主角三、消费者

MQ 服务：“我把消息给你了啊”

消费者：“接到了”

MQ 服务：“好，我把消息删了啊”

然后，消费者拿到信息之后，开始工作，突然～～～一命呜呼，任务没执行完，消息丢失了，应该完成的任务到此为止。

解决：消费者，消费完之后，回复ack ，确认已消费完。如果超时未回复，那么重新发放消息。也有可能一条消息，被多个消费者消费，这里业务代码要保证幂等性。

            $callback = function ($msg){


                //console log : Received message

                self::consoleLog(" [x] Received".$msg->body,0);


                //执行业务操作 (根据生产者，设定的路由 Http 访问)

                $data = json_decode($msg->body,true);

                if(isset($data['url']))

                {

                    $url = $data['url'].'?data='.$msg->body;

                    if(strpos($url,'http') !== false)

                    {

                        $result = file_get_contents($url);

                    }else{

                        $result = 'HTTP not found';

                    }

                    self::consoleLog($result);

                }else{

                    self::consoleLog('Undefined URL');

                }


                //手动 回复队列，message已消费

                $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);

            };


            //只有consumer已经处理，并确认了上一条message时queue才分派新的message给它(非公平分配，如果存在耗时操作，那么也一直等待。现在是空闲领取消息)

            self::$channel->basic_qos(null, 1, null);

            //第四个参数  是否自动回应 ack，false 手动回应

            self::$channel->basic_consume(self::$rabbit_queue,'',false,false,false,false,$callback);

三个演员的精湛表演，叹为观止，让他们同台演出，享受一番视觉盛宴吧！

1、ProducerClass  投递消息

    <?php

    namespace DemoQueue\Queue\Producer;

    /**

    * Created by PhpStorm.

    * User: runBaby

    * Date: 2019/5/13

    * Time: 11:13 AM

    */


    date_default_timezone_set("Asia/Shanghai");

    require_once __DIR__.'/../../vendor/autoload.php';


    use PhpAmqpLib\Wire\AMQPTable;

    use PhpAmqpLib\Message\AMQPMessage;

    use PhpAmqpLib\Connection\AMQPStreamConnection;


    class ProducerClass

    {

        static protected $rabbit_host;

        static protected $rabbit_port;

        static protected $rabbit_login;

        static protected $rabbit_pwd;

        static protected $rabbit_vhost;


        static protected $connection;

        static protected $channel;


        static protected $rabbit_err_code = 500;

        static protected $rabbit_success_code = 200;


        static protected $cache_exchange;

        static protected $cache_routing;

        static protected $cache_queue;


        static protected $config;


        const EXCHANGE_MODEL = 'fanout';                //交换机模式



        public function __construct($cache_exchange,$cache_queue)

        {

            self::$config = include_once __DIR__."/../Config/config.php";


            self::$config = self::$config['rabbitmq'];

            self::$rabbit_host = self::$config['host'];

            self::$rabbit_port = self::$config['port'];

            self::$rabbit_login = self::$config['login'];

            self::$rabbit_pwd = self::$config['password'];

            self::$rabbit_vhost = self:: $config['vhost'];


            self::$connection = new AMQPStreamConnection(self::$rabbit_host, self::$rabbit_port,self::$rabbit_login, self::$rabbit_pwd, self::$rabbit_vhost);


            if(!self::$connection->isConnected())

            {

                self::apiResponse(self::$rabbit_err_code,'建立连接失败');

            }


            self::$channel = self::$connection->channel();


            if(!self::$channel->is_open())

            {

                self::apiResponse(self::$rabbit_err_code,'通道连接失败');

            }


            if(!$cache_exchange)

            {

                self::apiResponse(self::$rabbit_err_code,'请设置交换机名称');

            }else{

                self::$cache_exchange = $cache_exchange;

            }


            if(!$cache_queue)

            {

                self::apiResponse(self::$rabbit_err_code,'请设置队列名称');

            }else{

                self::$cache_queue = $cache_queue;

                //路由 （同名 队列 借用）

                self::$cache_routing = self::$cache_queue;

            }

        }


        /**

        * Explain: 向队列 投递数据

        * @param array $send_info

        * User: runBaby

        * Date: 2019/5/13

        * Time: 11:34 AM

        * @return bool

        */

        public static function Producer($send_info = array())

        {

            //参数 json 转化

            $data = json_encode($send_info);


            //1、声明 交换机

            self::$channel->exchange_declare(self::$cache_exchange, self::EXCHANGE_MODEL, false, true, false);


            //2、声明队列

            self::$channel->queue_declare(self::$cache_queue, false, true, false, false, false);


            //3、队列绑定 交换机

            self::$channel->queue_bind(self::$cache_queue, self::$cache_exchange, self::$cache_routing);


            //4.1 设置异步回调消息确认 （生产者 防止信息丢失）

            self::$channel->set_ack_handler(

                function (AMQPMessage $message) {

                    echo "Message acked with content " . $message->body . PHP_EOL;

                    self::apiResponse(self::$rabbit_success_code, 'success', $message->body);

                }

            );

            self::$channel->set_nack_handler(

                function (AMQPMessage $message) {

                    echo "Message received failed，Please try again：" . $message->body . PHP_EOL;

                    self::apiResponse(self::$rabbit_err_code, 'Message received failed，Please try again', $message->body);

                }

            );


            //4.2 选择为 confirm 模式（此模式不可以和事务模式 兼容）

            self::$channel->confirm_select();


            //5、发送消息 到队列

            $msg = new AMQPMessage($data, array(

                //参数 发送消息的时候将消息的 deliveryMode 设置为 2 （RabbitMQ 持久化 步骤二：消息）

                'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT

            ));

            self::$channel->basic_publish($msg, self::$cache_exchange, self::$cache_queue);


            //4.3 阻塞等待消息确认 （生产者 防止信息丢失）

            self::$channel->wait_for_pending_acks();


            //请求相应 返回

            self::apiResponse(self::$rabbit_success_code, 'success', $data);


            return true;

        }



        /*

        * 资源返回

        */

        public static function apiResponse($code= 200 ,$message='默认描述信息',$data=[])

        {

            if(empty($data)){

                $data = (object)$data;

                self::producersLog($message);

            }else{

              self::producersLog($data,$message);

            }

            header('Content-Type:application/json; charset=utf-8');

            exit(json_encode(['code' => $code, 'message' => $message, 'data' => $data],JSON_UNESCAPED_UNICODE));

        }



        /*

        * 日志记录

        */

        public static function producersLog($data = array(),$message = '')

        {

            $filename = __DIR__.'/Log/producers/'.date('Y').'/'.date('m').'/'.date('Y-m-d').'.txt';


            $dir        =  dirname($filename);

            if(!is_dir($dir))

            {

                mkdir($dir,0777,true);

            }


            $log_str = '[ '.date('Y-m-d H:i:s').' ]'."\r\n";


            if($message){

                $log_str .= '*** message *** :'.$message."\r\n";

            }


            if(gettype($data) == 'array')

            {

                $data = json_encode($data);

            }

            $log_str .= $data."\r\n";

            $log_str .= '[end]'."\r\n";


            file_put_contents($filename,$log_str,FILE_APPEND);


            return true;

        }



        /*

        * 关闭连接

        */

        public function __destruct()

        {

            // TODO: Implement __destruct() method.

            self::$channel->close();

            self::$connection->close();

        }


    }

2、ConsumersClass  消费消息

    <?php

    namespace DemoQueue\Queue\Consumers;

    /**

    * Created by PhpStorm.

    * User: 奔跑吧笨笨

    * Date: 2019/5/6

    * Time: 1:04 PM

    */

    date_default_timezone_set("Asia/Shanghai");

    require_once __DIR__.'/../../vendor/autoload.php';


    use PhpAmqpLib\Connection\AMQPStreamConnection;


    class ConsumersClass

    {


      static protected $rabbit_host;

      static protected $rabbit_port;

      static protected $rabbit_login;

      static protected $rabbit_pwd;

      static protected $rabbit_vhost;


      static protected $connection;

      static protected $channel;


      static protected $config;


      static protected $rabbit_err_code = 500;

      static protected $rabbit_success_code = 200;


      static protected $rabbit_exchange;

      static protected $rabbit_queue;

      static protected $rabbit_routing;


        const EXCHANGE_MODEL = 'fanout';                //交换机模式 (广播模式)



        public function __construct($rabbit_exchange,$rabbit_queue)

        {

            self::$config = include_once __DIR__."/../Config/config.php";


            self::$config = self::$config['rabbitmq'];

            self::$rabbit_host = self::$config['host'];

            self::$rabbit_port = self::$config['port'];

            self::$rabbit_login = self::$config['login'];

            self::$rabbit_pwd = self::$config['password'];

            self::$rabbit_vhost =  self::$config['vhost'];


            self::$connection = new AMQPStreamConnection(self::$rabbit_host, self::$rabbit_port,self::$rabbit_login, self::$rabbit_pwd, self::$rabbit_vhost);


            if(!self::$connection->isConnected())

            {

                self::apiResponse(self::$rabbit_err_code,'建立连接失败');

            }


            self::$channel = self::$connection->channel();


            if(!self::$channel->is_open())

            {

                self::apiResponse(self::$rabbit_err_code,'通道连接失败');

            }


            if($rabbit_exchange)

            {

                self::$rabbit_exchange = $rabbit_exchange;

            }else{

                self::apiResponse(self::$rabbit_err_code,'请选择Exchange');

            }


            if($rabbit_queue)

            {

                self::$rabbit_queue = $rabbit_queue;

                //同名 借用 路由

                self::$rabbit_routing = $rabbit_queue;

            }else{

                self::apiResponse(self::$rabbit_err_code,'请选择Queue');

            }


        }



        /*

        * 消费者：客户端

        * 消费队列消息，并基于HTTP API路由转发到相应业务代码

        */

        public static function consumersClient()

        {

            //1、声明交换机

            self::$channel->exchange_declare(self::$rabbit_exchange, self::EXCHANGE_MODEL,false,true,false);

            //2、声明队列

            self::$channel->queue_declare(self::$rabbit_queue,false,true,false,false,false);

            //3、交换机和队列 绑定

            self::$channel->queue_bind(self::$rabbit_queue, self::$rabbit_exchange,self::$rabbit_routing);


            //console log : Start to work

            self::consoleLog();


            $callback = function ($msg){


                //console log : Received message

                self::consoleLog(" [x] Received".$msg->body,0);


                //执行业务操作 (根据生产者，设定的路由 Http 访问)

                $data = json_decode($msg->body,true);

                if(isset($data['url']))

                {

                    $url = $data['url'].'?data='.$msg->body;

                    if(strpos($url,'http') !== false)

                    {

                        $result = file_get_contents($url);

                    }else{

                        $result = 'HTTP not found';

                    }

                    self::consoleLog($result);

                }else{

                    self::consoleLog('Undefined URL');

                }


                //手动 回复队列，message已消费

                $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);

            };


            //只有consumer已经处理，并确认了上一条message时queue才分派新的message给它(非公平分配，如果存在耗时操作，那么也一直等待。现在是空闲领取消息)

            self::$channel->basic_qos(null, 1, null);

            //第四个参数  是否自动回应 ack，false 手动回应

            self::$channel->basic_consume(self::$rabbit_queue,'',false,false,false,false,$callback);



            //进入等待状态

            while (count(self::$channel->callbacks)) {

                self::$channel->wait();

            }


            return true;

        }



        /*

        * Console log

        */

        protected static function consoleLog($message = ' [*] Waiting for message. To exit press CTRL+C ',$type = 1)

        {

            $message = date('Y-m-d H:i:s').$message;


            //输出到 控制台

            echo $message.PHP_EOL;

            //是否记录文件日志

            if($type === 1)

            {

                self::Logs($message);

            }


            return true;

        }



        /*

        * API 返回

        */

      protected static  function apiResponse($code= 200 ,$message='默认描述信息',$data=[])

        {

            if(empty($data)){

                $data = (object)$data;

              self::Logs($message);

            }else{

                self::Logs($data);

            }

            header('Content-Type:application/json; charset=utf-8');

            exit(json_encode(['code' => $code, 'message' => $message, 'data' => $data],JSON_UNESCAPED_UNICODE));

        }



        /*

        * 日志记录

        */

      protected static function Logs($message = '')

        {

            $filename = __DIR__.'/Log/consumers/'.date('Y').'/'.date('m').'/'.date('Y-m-d').'.txt';


            $dir        =  dirname($filename);

            if(!file_exists($dir))

            {

                @mkdir($dir,0777,true);

            }


            $log_str = '[ '.date('Y-m-d H:i:s').' ]'."\r\n";

            $log_str .= '*** message *** :'.$message."\r\n";

            $log_str .= '[end]'."\r\n";


            file_put_contents($filename,$log_str,FILE_APPEND);


            return true;

        }



        /*

        * 销毁连接

        */

        public function __destruct()

        {

            // TODO: Implement __destruct() method.

            self::$channel->close();

            self::$connection->close();

        }


    }


3、PClient.php  执行调用class

    <?php

    namespace DemoQueue\Queue\Producer;

    /**

    * Created by PhpStorm.

    * User: runBaby

    * Date: 2019/5/13

    * Time: 11:37 AM

    */


    include_once __DIR__.'/ProducerClass.php';



    $data['type'] = 1;

    $data['data'] = 'Hello world!88888';


    $exchange = 'demo_exchange_test11';

    $queue = 'demo_queue_test11';


    $Producer = new ProducerClass($exchange,$queue);

    $result = $Producer::Producer($data);


    var_dump($result);

4、CClient.php  执行调用class

    <?php

    namespace DemoQueue\Queue\Consumers;

    /**

    * Created by PhpStorm.

    * User: runBaby

    * Date: 2019/5/6

    * Time: 1:41 PM

    */

    include_once __DIR__.'/ConsumersClass.php';



    $exchange = 'demo_exchange_test11';    //延迟交换机

    $queue = 'demo_queue_test11';          //延迟队列


    $consumers = new ConsumersClass($exchange,$queue);


    $consumers::consumersClient();

原文链接：https://blog.csdn.net/qq_37837134/java/article/details/90172003

©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 200,045评论 5赞 468
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 84,114评论 2赞 377
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 147,120评论 0赞 332
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 53,902评论 1赞 272
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 62,828评论 5赞 360
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 48,132评论 1赞 277
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 37,590评论 3赞 390
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,258评论 0赞 254
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 40,408评论 1赞 294
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,335评论 2赞 317
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,385评论 1赞 329
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,068评论 3赞 315
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 38,660评论 3赞 303
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,747评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 30,967评论 1赞 255
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 42,406评论 2赞 346
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 41,970评论 2赞 341