rabbitMQ安装及基础

Posted 2020-11-22 qiusanqi

参考文档:

1. 【rabbitmq-完整安装版】安装https://blog.csdn.net/csdn_azuo/article/details/100552423
2. php的amqp扩展 安装(windows) rabbitmq学习篇https://blog.csdn.net/weiwenjuan0923/article/details/79986951

时间轴:

1. 2019/10/31 安装centOS7虚拟机
2. 2019/11/04 linux下安装rabbitMQ服务
3. 2019/11/05 win下给安装php扩展amqp
4. 2019/11/12 win下php连接linus下rabbitMQ及生产者产生消息
5. 2019/11/13 解决页面访问消费者超时问题
6. 后续觉得还是看不懂博客举的例子和文档,在B站看了下牧码人入门视频(基础概念,实例较水)

 

 

问题:

1. 2019/11/05 php安装amqp扩展,Apache凉了, 扩展包下错了,PHP的architecture

Architecture

x86

2. 2019/11/12 Windows 下 PHP AMQP 链接错误 报错信息:Fatal error: Uncaught AMQPConnectionException: Socket error: could not connect to host. in E:PHPTutorialWWWindex.php:11 Stack trace: #0 E:PHPTutorialWWWindex.php(11): AMQPConnection->connect() #1 {main} thrown in E:PHPTutorialWWWindex.php on line 11(解决:网页使用15672端口,客户端连接使用5672端口, 5672端口未开启)

3. 2019/11/12 Deprecated: Function AMQPExchange::declare() is deprecated in E:PHPTutorialWWWindex.php on line 31 Exchange Status:1(解决:declare() 方法以及弃用,变为declareExchange())
4. 2019/11/13 消费者取数据时,响应时间很长,响应出现500错误 , 但是消息被领走
1. 错误信息: Internal Server Error

The server encountered an internal error or misconfiguration and was unable to complete your request.

Please contact the server administrator at [e-mail] to inform them of the time this error occurred, and the actions you performed just before this error.

More information about this error may be available in the server error log.

2. mq后台里队列有个消费者列表,在里面的话一般都能消费https://bbs.csdn.net/topics/392410948?list=9168275
3. 原因:页面使用Apache访问,Apache有响应时间
4. 解决:请求时通过cmd 进入php文件夹, php.exe 文件路径 ,来执行消费者文件
5. 这里需要注意的是这个方法：$queue->consume，queue对象有两个方法可用于取消息：consume和get。前者是阻塞的，无消息时会被挂起，适合循环中使用；后者则是非阻塞的，取消息时有则取，无则返回false。就是说用了consume之后，会同步阻塞，该程序常驻内存，不能用nginx，apache调用。 

 

涉及基础:

1.消息队列解决哪些问题

a.异步处理 : 提高系统执行速度

b.应用解耦 : 分布式(eg:超买超卖,解耦降低粘合度)

c.流量削锋 : 高并发(eg:抢红包,秒杀MQ进行过滤)

d.日志处理 : 大数据(eg:统计广告位数据,异步将数据日志存入文件->日志收集 flume单线程 td-agent单线程 logstash可开多进程->消息中间件->进行流计算 实时统计 || 归档后离线计算后数据清洗导入mysql后可出报表)

...

2.virtual hosts 管理

a.即vhosts,相当于mysql的db

b.一般以/开头

c.对用户进行授权

3.Overview->Ports and contents

协议	监听端口	备注
http	15672	自身服务器页面访问端口 浏览器与rabbitMQ 服务器通信使用http协议
amqp	5672	通信协议,与rabbitMQ服务器通信使用的协议
clustering	25672	支持集群

 

1.简单队列

P----Queue----C

a.三个对象: 生产者,队列,消费者 11对应

b.发布消息: 链接mq服务->创建通道->声明队列->通过channel发布消息到队列

c.消费消息: 链接mq服务->创建通道->定义队列的消费者channel->监听队列

d.不足: 耦合性高, 生产者一一对应消费者; 不能实现多个消费者消费队列中的消息; 队列名需要同时变更

 

2.工作队列

|----C1

P----Queue----|----C2

|----C3

a.模型: 一个生产者对应多个消费者

d.出现原因: 简单队列11对应; 实际开发中,生产者发送消息毫不费力, 而消费者一般是要跟业务相 结合的, 消费者接受到消息后需要处理,可能需要花费时间, 队列就可能挤压很多消息

c.生产者: 链接mq->创建通道->声明队列->发布消息->关闭通道

d.消费者: 链接mq->创建通道->通道声明队列->定义消费者接收消息

e.消费者1和2处理消息均分,1偶数2奇数数据,这种现象叫轮询分发,不管谁忙谁闲都不会多给消息,任务消息总是你一个我一个

 

3.公平分发

|----C1

P----Queue----|----C2

|----C3

a.使用公平分发必须关闭自动应答(ack)改成手动反馈

b.生产者: 每个消费者发送确认消息之前,消息队列不发送下一个消息到消费者,一次处理一个消息; 限制发送给同一个消费者不得超过一条消息

c.消费者2处理比消费者1多(能者多劳)

 

4.消息应答(autoAck布尔值,Message Acknowledge)

a.消息应答默认是打开的,默认为false

b. 为true时,自动确认模式, MQ将消息分发给消费者,该条消息就会将被从内存中消除

这种情况下,如果杀死正在执行的消费者,就会丢失正在处理的消息!

c. 为false时,手动模式,如果有一个消费者挂掉,就会交付给其他消费者.rabbitMQ支持消息应答,告诉rabbitMQ这个消息已经处理,可以删除了,rabbitMQ将删除内存中的消息,如果MQ服务器挂了,消息仍会丢失!

 

5.消息持久化(durable布尔型)

a.已定义过的队列修改durable属性,是不可以的;尽管代码正确,也不会运行成功

b.rabbitMQ不允许对已存在的队列进行重新定义不同的参数信息

 

6.订阅模式

　　　 |----Queue1----C1

P----x----|

|----Queue2----C2

a.模型解读:

一个生产者,多个消费者;

每一个消费者都有自己的队列;

生产者没直接把消息发送至队列,而是发送到了交换机(转发器exchange);

每个队列都要绑定带交换机上;

生产者发送的消息经过交换机到达队列,实现一个消息被多个消费者消费

b.应用场景:

注册后需要发邮件和短信等操作

商品修改后搜索引擎及前台都需要变更

c.生产者: 声明交换机->发布消息到交换机->关闭通道->关闭连接

交换机未绑定队列时,发送消息到队列会导致消息丢失,因为交换机没有存储的能力, 在rabbitMQ中只有队列有存储能力

d.消费者: 连接->创建channel->声明队列->绑定队列到交换机->定义消费者

 

7.Exchange(交换机 转发器)

a.匿名转发,exchange为空(""), 上述订阅模式

b.Fanout(不处理路由键): 消息经过交换机,与其交换机绑定的队列都能收到消息

c.Direct(处理路由键): 发送消息时带一个路由key,若绑定队列key与路由key相匹配,则将消息转指相应队列

 

8.路由模式

type=direct

| |---error---|---Queue1----C1

P----x--------|

|----info---|

|---error---|---Queue2----C2

|-warning-|

a.生产者:链接->创建通道->声明交换机->发布消息->关闭通道->关闭连接

b.消费者:链接->创建通道->声明队列->绑定交换机->定义消费者

 

9.话题模式

type=topic

| |---*.orange.*---|---Queue1----C1

P----x--------|

|----*.*.rabbit---|

|------lazy.#-----|---Queue2----C2

a.将路由和某个模式匹配,通配符

b.井号匹配一个或多个; 星号匹配一个

c.例如: Goods.# =(Goods.insert/delete)

d.场景: 生产者管理商品(增删改查)

 

10.消息确认机制(事务+confirm)

a.rabbitMQ中可以通过持久化数据解决rabbitMQ服务器异常的数据丢失问题

b.背景: 生产者将消息发送出去后,消息到底有没有到达rabbitMQ服务,默认情况下是不知道的

c.两种方式确定: AMQP实现了事务机制;Confirm模式

 

11.事务机制(AMQP本身协议自带):

a.txSelect(将当前channel设置成transaction模式);txCommit(用于提交事务);txRollback(回滚事务)

b.实现: 使用try catch抛出异常及事务机制实现确保消息发送至rabbitMQ服务器

c.不足: 协议通信次数太多,会降低消息的吞吐量且耗时,每个信息都要提交一次相当于一次请求

 

12.Confirm模式(事务模式队列不能重复设置)

a.生产者端confirm模式的实现原理: 生产者将信道设置成confirm模式,一旦信道进入confirm模式,所有在该信道上发布的消息都会被指派一个唯一的ID(从1开始),一旦消息被投递到所有匹配的队列之后,broker就会发送一个确认给生产者(包含消息的唯一ID),这就使生产者知道消息已经正确到达目的队列了,如果消息和队列时可持久化的,那么确认消息会将消息写入磁盘之后发出,broker回传给生产者的确认消息中deliver-tag域包含了确认消息的序列号,此外broker也可以设置basic.ack的multiple域,表示到这个序列号之前的所有消息已经得到了处理

b.confirm模式的最大好处在于其是异步的;

c.开启confirm模式: channel.confirmSelect();

d.编程模式:

普通 发一条 waitForConfirms(),等待服务端确认,串行方法;

批量 发一批 waitForConfirms();

异步 提供一个回调方法

Channel对象提供的ConfirmListener()回调方法只包含deliveryTag(当前Channel发出的消息序号),我们需要自己为每个Channel维护一个unconfirm的消息序号集合,每发布一条数据,集合中元素加1,每回调一次handleAck方法,unconfirm集合删掉相应的一条(multiple=false)或多条(multiple=true)记录.从程序运行效率上看,这个unconfirm集合最好采用有序集合SortedSet存储结构

生产者:链接->创建通道->声明队列->(通道添加监听)声明未确认的消息标识集合->发布成功后回调handleAck()函数,回执有问题执行handleNack()函数->关闭通道->关闭连接