rabbitmq系统学习

Posted 2021-01-22 sky-chen

各种mq

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

AMQP核心概念

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

急速入门

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

mq高级特性

1. 消息如何保证100%的投递成功
   • 生产端的可靠性投递
     • 保障消息的成功发出
     • 保障MQ节点的成功接收
     • 发送端收到MQ节点（Broker）确认应答
2. 可靠性投递
   • 消息落库，对消息状态进行打标
   • 消息的延迟投递，做二次确认，回调检查
3. 幂等性
   • 消息永远不会消费多次，就算收到多次，效果相同
   • 方案
     • 唯一ID+指纹码机制，利用数据库主键去重
     • 利用Redis的原子性去实现
4. Confirm确认消息
   • 消息的确认，是指生产者投递消息后，如果Broker收到消息，则会给我们生产者一个应答
   • 生产者进行接收应答，用来确定这条消息是否正常的发送到Broker，这种方式也是消息的可靠性投递的核心保障
5. Return消息机制
   • Return Listener用于处理一些不可路由的消息
   • 我们的消息生产者，通过制定一个Exchange和Routingkey，把消息送达到某个队列中去，然后我们的消费者监听队列，进行消费处理操作
   • Mandatory：如果为true，则监听器会接收到路由不可达的消息，然后进行后续处理，如果为false，那么broker端自动删除该消息
6. 消费端自定义监听
   • 继承DefaultConsumer
   • 实现handleDelivery方法，构造函数传入channel
7. 消费端限流
   • 例子：假设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条都应答了才继续推送
8. 消费端ACK与重回队列
   • 消费端的手工ACK和NACK
   • 消费端进行消费的时候，如果由于业务异常我们进行日志的记录，然后进行补偿
   • 如果由于服务器宕机等严重问题，那我们就需要进行ACK保障消费端消费成功
   • 一般我们在实际应用中，都会关闭重回队列，也就是设置为false
   • 在应答的时候，设置是否重回队列队尾
9. TTL队列/消息
   • Time To Live的缩写，也就是生存时间
   • RabbitMQ支持消息的过期时间，在消息发送时可以进行制定
   • RabbitMQ支持队列的过期时间，从消息入队列开始计算，只要超过了队列的超时时间配置，那么消息会自动的清除
10. 死信队列
    • 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);