zoukankan      html  css  js  c++  java
  • laravel框架的rabbitmq使用示例[多队列封装]

        RabbitMQ是实现了高级消息队列协议(AMQP)的开源消息代理软件(亦称面向消息的中间件)。RabbitMQ服务器是用Erlang语言编写的,而集群和故障转移是构建在开放电信平台框架上的。所有主要的编程语言均有与代理接口通讯的客户端库。

        消息队列工作示意图

        

         生产者P--[发布消息]--> 交换机X--[根据路由绑定分发]-->队列<--[订阅消息]--消费者C

         关系

        1 生产者和交换机

             创建消息,并且发送到对应的交换机,发送的时候可以带上特定的routeKey

        2 交换机和队列

           队列 绑定到 交换机  (可以设置路由 routeKey  direct 和 topic 模式下 有效,广播模式只要绑定就分发到队列)

           假如交换机上的消息分发不到队列,则此消息就自动删除了

           a 直连交换机(Direct exchange)

               交换机 --[所有绑定在自己上的队列中找出设置和消息routeKey一样的]--> 队列 ,根据绑定的routeKey 来找队列 分发消息
           b 广播交换机 (Fanout Exchange)
               交换机 --[所有绑定在自己上的]--> 队列,  只要队列绑定到交换机 队列分发消息
           c 主题交换机 (Topic Exchange)
               交换机 ---[所有绑定在自己上的队列中找出 消息routeKey 满足队列匹配的]--> 队列, routeKey 满足匹配要求的队列就会分发消息.

         3 队列和消费者

            消费者绑定到对应的队列就能得到队列中的消息,  假如多位消费者同事消费一个队列 可以通过 prefetchCount 来设置 最多同时消费个数, 握手后再发送新的消息过来

        示例代码说明

        消息队列虽然是持久化,可以通过握手机制来实现是否正真消费。示例代码中采用了默认握手,通过数据库记录中存放对应执行记录来实现队列的执行情况监控。

        1 rabbitmq操作

            新建demo 账号
            rabbitmqctl add_user demo 181219

            新建demo 虚拟主机
            rabbitmqctl add_vhost demo

            设置 demo 账号在 demo 虚拟主机 权限
            rabbitmqctl set_permissions -p demo demo ".*" ".*" ".*"


            web界面插件开启
            rabbitmq-plugins enable rabbitmq_management

            设置demo 账号 角色
            rabbitmqctl set_user_tags demo administrator

         2 数据库表

        表一共2张 一张记录消息内容以及执行情况, 另一张记录执行失败的错误信息

    CREATE TABLE `mq_process_error_log` (
      `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT,
      `msg_id` bigint(20) NOT NULL,
      `process_msg` varchar(255) NOT NULL DEFAULT '' COMMENT '执行返回信息',
      `create_time` int(10) NOT NULL DEFAULT '0' COMMENT '创建时间',
      PRIMARY KEY (`id`),
      KEY `idx_msg_id` (`msg_id`) USING BTREE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COMMENT='消息处理失败日志表';
    
    -- ----------------------------
    -- Table structure for mq_process_log
    -- ----------------------------
    CREATE TABLE `mq_process_log` (
      `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT,
      `msg_str` varchar(200) NOT NULL DEFAULT '' COMMENT '消息请求内容 json字符串',
      `msg_type` tinyint(2) NOT NULL DEFAULT '0' COMMENT '消息类型',
      `find_keyword` varchar(32) NOT NULL DEFAULT '' COMMENT '查找消息内容的关键字',
      `create_time` int(10) NOT NULL DEFAULT '0' COMMENT '创建时间',
      `process_num` tinyint(2) NOT NULL DEFAULT '0' COMMENT '执行次数',
      `process_start_time` int(10) NOT NULL DEFAULT '0' COMMENT '执行开始时间',
      `process_end_time` int(10) NOT NULL DEFAULT '0' COMMENT '执行结束时间',
      `process_status` tinyint(1) NOT NULL DEFAULT '0' COMMENT '执行状态 0 未执行 1 成功 2 失败',
      `process_msg` varchar(255) NOT NULL DEFAULT '' COMMENT '执行返回信息',
      PRIMARY KEY (`id`),
      KEY `idx_find_keyword` (`find_keyword`) USING BTREE,
      KEY `idx_msg_type` (`msg_type`) USING BTREE,
      KEY `idx_process_status` (`process_status`) USING BTREE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COMMENT='消息处理日志表';

         3 代码说明

           创建3个队列来处理不同的消息, 来满足不同优先级消息处理

           发送消息,以及处理消息监听,重发消息都有封装

          代码中已经创建了抽象父类,不同的vhost配置只需要继承父类即可, 如下代码

         

    /**
     * demo-rabbitMq类
     * Class DemoRabbitMq
     * @package AppServiceAmqp
     * @author zxqc2018
     */
    class DemoRabbitMq extends AbstractRabbitMq
    {
        use Singleton;
        //mq配置
        protected $configName = 'demo';
        //默认交换机
        protected $defaultExchangeName = 'demo-exchange';
        //快中慢-队列配置
        protected $queuePriorityConfig = [
            'fast' => ['demo-fast-queue', 'demo.fast.#'],
            'middle' => ['demo-middle-queue', 'demo.middle.#'],
            'slow' => ['demo-slow-queue', 'demo.slow.#'],
        ];
    
        /**
         * 设置routeKey对应处理方法
         * @author zxqc2018
         */
        function settingRouteKeyProcessFunc()
        {
            $this->routeKeyProcessFunc[self::DEMO_FAST_TEST] = function ($msgData) {
                print_r($msgData);
                $res = 
    esultData();
                $res->setMessage('fast');
                return $res;
            };
            $this->routeKeyProcessFunc[self::DEMO_SLOW_TEST] = function ($msgData) {
                print_r($msgData);
                if (rand(1,10) > 5) {
                    return resultData([], ErrorCode::ERROR_RABBIT_MQ, '测试处理失败');
                }
                $res = 
    esultData();
                $res->setMessage('slow');
                return $res;
            };
        }
    }

         代码地址:

         https://gitee.com/zxqc2014/laravel_rabbitmq_demo

    作者:做想其成
    本文版权归作者和博客园共有,欢迎转载,但未经作者同意必须在文章页面给出原文链接,否则保留追究法律责任的权利。
  • 相关阅读:
    文本切换器(TextSwitcher)的功能和用法
    图像切换器(ImageSwitcer)的功能与用法
    ViewSwitcher的功能与用法
    HTTP协议-get请求与post请求的区别
    HTTP协议缓存
    HTTP协议详解
    Vue.js----更换头像不实时更新问题
    Vue.js----date与时间戳的转换(unixTime)Moment.js让日期处理变得更简单
    Let's Encrypt 免费通配 https 签名证书 安装方法
    小程序Openid 获取,服务器 encryptedData 解密 遇到的坑
  • 原文地址:https://www.cnblogs.com/zxqc/p/10684078.html
Copyright © 2011-2022 走看看