rabbitmq系统学习（一）

老梁

发布于 2019-09-10 17:42:31

7650

发布于 2019-09-10 17:42:31

文章被收录于专栏：Java工程师成长之路Java工程师成长之路

各种mq

activemq，kafka使用zookeeper做管理
rocketmq自己实现nameserver broke管理

AMQP核心概念

高级消息队列协议
publisher application->Server->Virtual host->Exchange->Message Queue->Consumer application
Server:又称Broker，接收客户端的连接，实现AMQP实体服务
Connection：连接，应用程序与Broker的网络连接
Channel：网络信道，几乎所有操作都在Channel中进行，Channel是进行消息读写的通道。客户端可建立多个Channel，每个Channel代表一个会话任务
Message：消息，传递的数据，由Properties和Body组成。Properties可以对消息进行修饰，比如消息的优先级、延迟等高级特性；Body则是消息体内容
Virtual host：虚拟地址，用于进行逻辑隔离，最上层的消息路由。一个Virtual Host里面可以有若干个Exchange和Queue，同一个Virtual Host里面不能有相同名称的Exchange或Queue
Exchange：交换机，接收消息，根据路由键转发消息到绑定的队列
Binding：Exchange和Queue之间的虚拟连接，binging中可以包含routing key
Routing key：一个路由规则，虚拟机可以用它来确定如何路由一个特定消息
Queue：也称为Message Queue，消息队列，保存消息并将它们转发给消费者

急速入门

消费端
- durable true表示服务重启也不会删除消息
- exclusive true表示独占一个队列，保证顺序消费
- autoDelete true表示绑定接触，自动删除交换机
Exchange：
- Auto Delete：当最后一个绑定到Exchange上的队列删除后，自动删除该Exchange
- Internal：当前Exchange是否用于RabbitMQ内部使用，默认为False
Direct Exchange
- Direct模式采用RabbitMQ自带的Exchange：default Exchange，所以不需要讲Exchange进行任何绑定binding操作，消息传递时，RouteKey必须完全匹配才会被队列接收，否则该消息会被抛弃
- 这种模式常用语单一队列
Topic Exchange
- 所有发送到Topic Exchange的消息被转发到所有关心RouteKey中指定Topic的Queue上
- Exchange将RouteKey和某Topic进行模糊匹配，此时队列需要绑定一个Topic
- 通配符
  - # 匹配一个或多个词
  - * 匹配不多不少一个词
Fanout Exchange
- 不处理路由键，只需要简单的将队列绑定到交换机上
- 发送到交换机的消息都会被转发到与该交换机绑定的所有队列上
- Fanout交换机转发消息是最快的
Bingding-绑定
- Exchange和Exchange、QUeue之间的连接关系
Queue-消息队列
- 消息队列，实际存储消息数据
- Durability：是否持久化，Durable 是 Transient 否
- Auto delete：如果yes，当最后的监听被移除，该Queue会自动被删除
Message-消息
- 服务器和应用程序之间传送的数据
- 本质上就是一段数据，由Properties和Payload（Body）组成
- 常用属性：delivery mode、headers（自定义属性）
Message-其他属性
- content_type、content_encoding、priority
- correlation_id、reply_to、expiration、message_id
Virtual host-虚拟主机
- 虚拟地址，用于进行逻辑隔离，最上层的消息路由
- 一个Virtual Host里面可以有若干个Exchange和Queue
- 同一个Virtual Host里面不能有相同名称的Exchange或Queue

mq高级特性

消息如何保证100%的投递成功
- 生产端的可靠性投递
  - 保障消息的成功发出
  - 保障MQ节点的成功接收
  - 发送端收到MQ节点（Broker）确认应答
可靠性投递
- 消息落库，对消息状态进行打标
- 消息的延迟投递，做二次确认，回调检查
幂等性
- 消息永远不会消费多次，就算收到多次，效果相同
- 方案
  - 唯一ID+指纹码机制，利用数据库主键去重
  - 利用Redis的原子性去实现
Confirm确认消息
- 消息的确认，是指生产者投递消息后，如果Broker收到消息，则会给我们生产者一个应答
- 生产者进行接收应答，用来确定这条消息是否正常的发送到Broker，这种方式也是消息的可靠性投递的核心保障
Return消息机制
- Return Listener用于处理一些不可路由的消息
- 我们的消息生产者，通过制定一个Exchange和Routingkey，把消息送达到某个队列中去，然后我们的消费者监听队列，进行消费处理操作
- Mandatory：如果为true，则监听器会接收到路由不可达的消息，然后进行后续处理，如果为false，那么broker端自动删除该消息
消费端自定义监听
- 继承DefaultConsumer
- 实现handleDelivery方法，构造函数传入channel
消费端限流
- 例子：假设Rabbitmq服务器有上万未处理的消息，我们打开一个消费者客户端，会出现下面情况：
  - 巨量的消息瞬间全部推送过来，但是我们单个客户端无法同时处理这么多数据
- Rabbitmq提供了一种qos（服务质量保证）功能，即在非自动确认消息的前提下，如果一定数目的消息（通过基于consume或者channel设置Qos的值）未被确认前，不进行消费新的消息
- void BasicQos(uint prefetchSize,ushort prefetchCount,bool global);
- 消费者消费量限制方法
- prefetchCount：表示不要同时给消费者推送多余N个消息，即一旦有N个消息还没有ack，则该consumer将block掉，知道有消息ack
- global：true/false 是否将上面设置应用于channel，简单说，就是上面限制是channel级别还是consumer级别
- prefetchSize和global这两项，rabbitmq没有实现，prefetch_count在no_ask=false的情况下生效，即在自动应答的情况下这两个值是不生效的
- autoAck设置为false
- channel.basicQos(0,3,false);//表示consumer级别，限制3条都应答了才继续推送
消费端ACK与重回队列
- 消费端的手工ACK和NACK
- 消费端进行消费的时候，如果由于业务异常我们进行日志的记录，然后进行补偿
- 如果由于服务器宕机等严重问题，那我们就需要进行ACK保障消费端消费成功
- 一般我们在实际应用中，都会关闭重回队列，也就是设置为false
- 在应答的时候，设置是否重回队列队尾
TTL队列/消息
- Time To Live的缩写，也就是生存时间
- RabbitMQ支持消息的过期时间，在消息发送时可以进行制定
- RabbitMQ支持队列的过期时间，从消息入队列开始计算，只要超过了队列的超时时间配置，那么消息会自动的清除
死信队列
- DLX，Dead-Letter-Exchange
- 利用DLX，当消息在一个队列中变成死信（dead message）之后，它能被重新publish到另一个Exchange，这个Exchange就是DLX
- 消息被拒绝（basic.reject/basic.nack）并且requeuefalse
- 消息TTL过期
- 队列达到最大长度
- DLX也是一个正常的Exchange，和一般的Exchange没有却别，它能在任何的队列上被指定，实际上就是设置某个队列的属性
- 当这个队列中有死信时，RabbitMQ就会自动的将这个消息重新发布到设置的Exchange上去，进而被路由到另一个队列
- 可以监听这个队列中的消息做相应的处理，这个特性可以弥补RabbitMQ3.0以前支持的immediate参数的功能
- 使用
  - 正常的绑定
  - 然后需要在队列上加上一个参数:arguments.put("x-dead-letter-exchange","dlx.exchange");
- agruments放在声明的队列上
- channel.queueDeclare(queueName,true,false,false,agruments);