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() unsigned NOT NULL AUTO_INCREMENT,

  `msg_id` bigint() NOT NULL,

  `process_msg` varchar() NOT NULL DEFAULT '' COMMENT '执行返回信息',

  `create_time` int() NOT NULL DEFAULT '' 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() unsigned NOT NULL AUTO_INCREMENT,

  `msg_str` varchar() NOT NULL DEFAULT '' COMMENT '消息请求内容 json字符串',

  `msg_type` tinyint() NOT NULL DEFAULT '' COMMENT '消息类型',

  `find_keyword` varchar() NOT NULL DEFAULT '' COMMENT '查找消息内容的关键字',

  `create_time` int() NOT NULL DEFAULT '' COMMENT '创建时间',

  `process_num` tinyint() NOT NULL DEFAULT '' COMMENT '执行次数',

  `process_start_time` int() NOT NULL DEFAULT '' COMMENT '执行开始时间',

  `process_end_time` int() NOT NULL DEFAULT '' COMMENT '执行结束时间',

  `process_status` tinyint() NOT NULL DEFAULT '' COMMENT '执行状态 0 未执行 1 成功 2 失败',

  `process_msg` varchar() 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 App\Service\Amqp

 * @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 = \resultData();

            $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 = \resultData();

            $res->setMessage('slow');

            return $res;

        };

    }

}

代码地址:

https://gitee.com/zxqc2014/laravel_rabbitmq_demo

秒客网

laravel框架的rabbitmq使用示例[多队列封装]

相关文章