rabbitmq系统学习(一)

各种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);

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

扫码关注云+社区

领取腾讯云代金券