RabbitMQ高级特性实战

点击名片！

关注我吧！

哈喽大家好！我是小三，今天写的是关于Rabbitmq的高级特性。如果对你有帮助也可以点点关注点点赞哈。

RabbitMQ高级特性-消息可靠性的投递+消费

那什么是消息的可靠性投递呢？不知道小伙伴们有没有考虑过，生产者将数据将数据发送到Rabbitmq的时候，消息可能会因为网络等等一些问题在传入的过程里弄丢了。

先来看一张简单的消费流程图：

在上图可以想到：

1、从producker在往broker这一阶段的消息在传入的过程中会丢失。

2、消息发送到mq后，存入内存但还没来得及处理消息，因为宕机导致了内存中的数据丢失。

3、消费消费到数据之后，还没来得及处理，消费的进程就已经挂掉了，导致了mq以为消息已经被处理，事实上并没有。

因为上述原因，我们需要保证到mq节点可以成功的接收到消息，那怎么来保证消息可靠性呢？

我们知道rabbitmq的消息投递路径是：生产者->交换机->队列->消费者，通过这个消息投递路径我们可以通过两个点的控制来保证消息的可靠性传递。

1、生产者到交换机，通过confirmCallback，也就是确认回调，默认的话是不开启的。当这个模式开启之后，生产者给交换机发送一个消息，交换机收到消息后会给消费者发送一个确认收到的信息

2、交换机到队列，通过returnCallback，也就是返回回调，这个不像上一个的机制，confirmCallback不管消息是否成功到达交换机的时候都会被调用，而returnCallback只有在交换机到达队列失败的时候才会触发，当回调的函数被调用的时候，说明了交换机的消息并没有顺利的到达队列。

这里给个小建议：开启了消息的确认机制之后，是为了保证消息的准确送达，但会因为频繁的确认交互，使得rabbitmq的整体效率变得低下，吞吐量会下降严重。所以不是非常重要的消息真心不建议使用消息确认的机制

Rabbitmq消息可靠性confirmCallback实战

生产者到交换机，生产者投递消息之后，如果Broker接收到消息后，就给生产者发个ACK，ACK就好像是回馈消息，生产者就可以通过这一个ACK就可以确认这条消息是否正常地发送到Broker了，这种方式是消息可靠投递的核心。

开启confirmCallback的方法

在旧版中，确认消息的发送成功，可以通过实现ConfirmCallBack接口，把消息发送到交换器Exchange后触发回调

spring.rabbitmq.publisher-confirms=true


在新版里，NONE值已经禁用了发布确认模式，为默认值；CORRELATED的值是发布消息成功后到交换器会触发回调方法

spring.rabbitmq.publisher-confirm-type:correlated
实战演练@Autowired
private RabbitTemplate template;
@Test
void testConfirmCallback() {
template.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {

* @param correlationData 配置
* @param ack 交换机是否收到消息，true是成功，false是失败
* @param cause 失败的原因
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
System.out.println("confirm=====>");
System.out.println("confirm====ack="+ack);
System.out.println("confirm====cause="+cause);
//根据ACK状态做对应的消息更新操作 TODO


template.convertAndSend(RabbitMQConfig.EXCHAN
GE_NAME+,"order.new","新订单来啦1");
}


如果需要模拟异常的话需要对投递的交换机每次进行更改！简单的实战就到这里了

RabbitMQ死信队列+TTL TTL是什么？

TTL的意思就是time to live 消息的存活时间。如果消息一直在队列当中没有被消费并且的话存在的时间已经超过了消息的存活时间，消息就会标称了“死信”（后续会讲到），后续就无法被消费了。设置TTL的话有两种方式：第一种单独消息来进行配置ttl，第二种是整个队列来进行配置ttl，第二种使用居多。TTL介绍完了，咱们来介绍一下死信队列

什么是死信队列呢？

可以从概念上来搞清楚，“死信”顾名思义就是无法被消费到的消息。一般来说的话，producer将消息投递到了broker或者是queue里面了，再接着consumer从queue取出消息后进行消费。这个没问题吧，但是在某一些时候由于特殊的原因会导致到queue中的某一些消息没有办法被消费到，这样的消息如果没有经过后续的处理的话，就会变成了死信，有了死信也就有了死信队列。

rabbitmq产生死信会有如下几种原因：

• 消息被拒绝了，并且requeue为false

• 消息的存活时间过期

• 队列达到的最大的长度（队列已经满了，没有办法在添加数据到mq里）

当消息成为死信之后，为了防止消息的丢失，会将这样无法处理到的消息发送到一个叫死信交换机里，后面再通过死信交换机绑定的路由键发送到相对应的死信队列。当消息在一个队列当中变成死信，如果也配置了死信队列，那么就会重新的publish到死信交换机中，私信交换机在把这些死信投递到队列上，就形成了死信队列。

在RabbitMQ管控台消息TTL进行测试

设置队列过期使用的参数，对整个队列的消息统一设置过期 x-message-ttl，消息过期时间使用的参数（如果队列的头部消息没有过期，队列中消息就已经过期了，已经还在队列里）expiration

RabbitMQ的Web控制台测试

新建死信交换机，这个和普通的没啥区别

新建死信队列

把死信交换机和队列进行绑定

新建普通的队列，并设置过期时间与指定的死信交换机

vhost可以认为是一个虚拟的小型版rabbitmq队列，它的内部有含有独立的queue、exchange和binding等等，还拥有独立的权限系统，可以做到vhost的范围用户控制，更多用户做不同的权限隔离。在用于不同的业务模块的逻辑隔离，一个Virtual Host里面可以有一个或者是多个Exchange和Queue，同一个VirtualHost的话里面不可以有相同名称的Exchange或者是Queue

2.面试官：在项目里，如何选择哪一个队列产品呢？

主流的消息队列有：ActiveMQ、Kafka、RabbitMQ、RocketMQ

ActiveMQ：Apache出品，历史悠久，⽀持多种语⾔的客户端和协议，⽀持多种语⾔Java, .NET, C++ 等，基于JMS Provider的实现

缺点：吞吐量不⾼，多队列的时候性能下降，存在消息丢失的情况，⽐较少⼤规模使⽤

Kafka：是由Apache软件基⾦会开发的⼀个开源流处理平台，由Scala和Java编写。Kafka是⼀种⾼吞吐量的分

布式发布订阅消息系统，它可以处理⼤规模的⽹站中的所有动作流数据(⽹⻚浏览，搜索和其他⽤户的⾏动)，副本集机制，实现数据冗余，保障数据尽量不丢失；⽀持多个⽣产者和消费者类似MQ，功能较为简单，主要⽀持简单的MQ功能

缺点：不⽀持批量和⼴播消息，运维难度⼤，⽂档⽐较少, 需要掌握Scala

RocketMQ：阿⾥开源的⼀款的消息中间件, 纯Java开发，具有⾼吞吐量、⾼可⽤性、适合⼤规模分布式系统应⽤的

特点, 性能强劲(零拷⻉技术)，⽀持海量堆积, ⽀持指定次数和时间间隔的失败消息重发,⽀持consumer端tag过滤、延迟消息等，在阿⾥内部进⾏⼤规模使⽤，适合在电商，互联⽹⾦融等领域，基于JMS Provider的实现

缺点：社区相对不活跃，更新⽐较快，纯java⽀持

RabbitMQ：是⼀个开源的AMQP实现，服务器端⽤Erlang语⾔编写，⽀持多种客户端，如：Python、Ruby、.NET、Java、C、⽤于在分布式系统中存储转发消息，在易⽤性、扩展性、⾼可⽤性等⽅⾯表现不错

缺点：使⽤Erlang开发，阅读和修改源码难度⼤

3.面试官：讲解下怎么样可以避免重复消费

任何消息队列产品不保证消息不重复，如果你的业务需要保证严格的不重复消息，需要你⾃⼰在业务端去重，kafka、rocketmq、rabbitmq等都是⼀样的，接⼝幂等性保障 ，消费端处理业务消息要保持幂等性。

「使用redis」

setNX

//Redis中操作，判断是否已经操作过
TODO
boolean flag = jedis.setNX(key);
if(flag){
//消费
}else{
//忽略，重复消费


Incr原子操作：key⾃增，⼤于0 返回值⼤于0则说明消费过

int num = jedis.incr(key);
if(num == 1){
//消费
}else{
//忽略，重复消费


上述两个⽅式都可以，但是排重可以不考虑原⼦问题，数据量多需要设置过期时间，考虑原⼦问题

END