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; }; } }
代码地址: