RabbitMQ实战应用技巧

共计 5284 个字符，预计需要花费 14 分钟才能阅读完成。

由于项目原因，之后会和 RabbitMQ 比较多的打交道，所以让我们来好好整理下 RabbitMQ 的应用实战技巧，尽量避免日后的采坑

RabbitMQ 有几个重要的概念：虚拟主机，交换机，队列和绑定

虚拟主机：一个虚拟主机持有一组交换机、队列和绑定，我们可以从虚拟主机层面的颗粒度进行权限控制
交换机：Exchange 用于转发消息，它并不存储消息，如果没有 Queue 队列绑定到 Exchange，它会直接丢弃掉生产者发来的数据。

交换机还有个关联的重要概念：路由键，消息转发到哪个队列根据路由键决定

绑定：就是绑定交换机和队列，它是 多对多的关系，也就是说多个交换机可以绑同一个队列，也可以一个交换机绑多个队列

交换机有四种类型的模式Direct, topic, Headers and Fanout

Direct 模式使用的是 RabbitMQ 的默认交换机，也是最简单的模式，适合比较简单的场景

如下图所示，使用 Direct 模式，我们需要创建不同的队列，而默认交换机则通过 Routing key 路由键的值来决定转发到哪个队列，可以看到，路由键绑定队列是可以指定多个的

Topic 模式主要是根据通配符匹配，也就类似于模糊匹配，当这种匹配模式和路由键匹配后交换机就能转发消息到指定队列

路由键为一串字符串，由句号（.）隔开，比如a.b.c
（*）代表指定位置一个单词，（#）代表零个或者多个单词，比如a.*.b.#，表示 a 和 b 中间随意填个单词，b 后面可以跟 n 个单词，比如a.x.b.c.d.e

Topic 模式和 Direct 模式的区别在于交换机需要自己指定，路由键支持模糊匹配，例如：

rabbitTemplate.convertAndSend("topicExchange","a.x.b.d", "hello world!");

Headers 也是根据规则匹配，但它不是根据路由键了，headers 有个自定义匹配规则，它将匹配键值设在了消息的 headers 属性上，当这些键值对有一对或者全部匹配时，消息才会被投递到对应队列，这种模式效率相对较低，一般不推荐使用

Fanout 即为大名鼎鼎的广播模式了，它 不需要管路由键 ，会把消息发给绑定它的全部队列，就算配置了 路由键也会被忽略

首先我们 Direct 模式，一个生产者一个消费者的情况，也就对应了一个发送者和一个队列 A 接收，这是没有疑问的，发送什么接收什么
当 Direct 模式，一个生产者发消息，开启多个消费者也就是多个相同 queue，此时消息由多个消费者均匀分摊，不会重复消费（前提 ack 正常）
当 Topic 模式，一个交换机绑定两个队列，路由键有重叠关系，如下代码，此时指定路由键 topic.message 发送消息，队列 queueMessage 和queueMessages都能接收到相同消息，也就是说，topic 模式可以实现类似于广播模式的形式，甚至更加灵活，它能否转发到消息由路由键决定。
相比于 Fanout 模式，我们如果要对消费者队列分组发送，我们需要指定不同的 路由键；而 Fanout 模式则需要指定不同的交换机和队列绑定，实际使用结合实际情况

@Configuration
public class TopicRabbitConfig {

    final static String message = "topic.message";
    final static String messages = "topic.messages";

    @Bean
    public Queue queueMessage() {return new Queue(TopicRabbitConfig.message);
    }

    @Bean
    public Queue queueMessages() {return new Queue(TopicRabbitConfig.messages);
    }

    @Bean
    TopicExchange exchange() {return new TopicExchange("exchange");
    }

    @Bean
    Binding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");
    }

    @Bean
    Binding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) {return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");
    }
}

我们的常用配置如下

spring.rabbitmq.addresses=localhost:5672
spring.rabbitmq.username=user
spring.rabbitmq.password=123456
spring.rabbitmq.virtual-host=/
spring.rabbitmq.connection-timeout=1000
## 设置监听限制：最大 10，默认 5
spring.rabbitmq.listener.simple.concurrency=5
spring.rabbitmq.listener.simple.max-concurrency=10
spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.publisher-returns=true
spring.rabbitmq.template.mandatory=true
spring.rabbitmq.listener.simple.acknowledge-mode=manual

其中最后四条配置需要着重解释：

spring.rabbitmq.publisher-confirms为 true，表示生产者消息发出后，MQ 的 broker 接收到了消息，发送回执表示确认接收，不设置则可能导致消息丢失
spring.rabbitmq.publisher-returns为 true，表示当消息不能到达 MQ 的 Broker 端，，则使用监听器对不可达的消息做后续处理，这种一般是路由键没配好，或 MQ 宕机才可能发生
spring.rabbitmq.template.mandatory当上面两个为 true 时，这个一定要配 true，否则上面两个不起作用
spring.rabbitmq.listener.simple.acknowledge-mode这个为 manual 表示手工确认，实际生产应该设为手工，才能保证你的业务是处理完成的，注意业务的幂等性，可重复调用，手工确认代码如下例子

@Component
public class RabbitReceiver {

    
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "queue-1", 
            durable="true"),
            exchange = @Exchange(value = "exchange-1", 
            durable="true", 
            type= "topic", 
            ignoreDeclarationExceptions = "true"),
            key = "springboot.*"
            )
    )
    @RabbitHandler
    public void onMessage(Message message, Channel channel) throws Exception {System.err.println("--------------------------------------");
        System.err.println("消费端 Payload:" + message.getPayload());
        Long deliveryTag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
        // 手工 ACK, 获取 deliveryTag
        channel.basicAck(deliveryTag, false);
    }
}

queue：队列的名字
durable：为 true 表示队列中数据持久化到磁盘，可以防止 mq 宕机重启数据丢失
exclusive：为 true 表示排他性，只允许一个当前连接访问该队列，当前已连接就不允许新的连接进入否则报错，当连接断开当前队列会销毁
autoDelete：为 true 表示自动删除，当没有 Connection 连接到队列的时候，会自动删除
arguments：这个参数用来添加一些额外参数的，如下图片
- 比如添加 x-message-ttl 为 5000，则表示消息超过 5 秒没被处理就会超时过期；
- x-expires设置 120000 表示队列在 2 分钟内没被消费则被删除；
- x-max-length,x-max-length-bytes表示传送数据的最大长度和字节数
- x-dead-letter-exchange，x-dead-letter-routing-key表示死信交换机和死信路由，放在需要过期或处理失败的队列属性中，这些数据会转发到死信队列存储起来，创建普通的交换机和队列绑定，把交换机名填到 x-dead-letter-exchange 的值，填写路由键要符合死信队列的路由键
- x-max-priority，表示设置优先级，范围为 0~255，只有 当消息堆积的时候，这个优先级才有意义，数字越大优先级越高
- x-queue-mode当为lazy，表示惰性队列，3.6.0 之后才被引入的概念，相比默认的模式，惰性队列模式会将生产者产生的消息直接存到磁盘中，这当然会增加 IO 开销，但适合应对大量消息堆积的情况；因为当大量消息堆积时，内存也不够存放，会将消息转存到磁盘，这个过程也是比较耗时且过程中不能接收新的消息。如果需要将普通队列转换成惰性队列需要将原来的队列删除，重新创建个惰性队列绑定。

![](https://image-static.segmentfault.com/406/935/4069353622-5db7b891b3bb4_articlex)

exchange：交换机名称
type：交换机类型
durable：持久化，同队列
autoDelete：是否自动删除，同队列
internal：若为 true，表示这个 exchange 不可以被 client 用来推送消息，仅用来进行 exchange 和 exchange 之间的绑定。
arguments：额外参数，目前只有个 alternate-exchange，表示当生产者发送消息到这个交换机，路由不到该交换机的队列，则会尝试这个参数指定的交换机进行路由，若路由键匹配，则路由到alternate-exchange 指定的队列，相当于转发了，刚好和上一个参数 internal 配合，若不想本交换机起到路由队列的作用，可以设置 internal 为 true，把消息都转发到 alternate-exchange 指定的交换机，由该交换机来路由指定队列，
- 如下图：exchange0设置了 alternate-exchange 交换机为 exchange1，生产者发送数据到exchange0 路由键为 test1，在exchange0 路由不到，则转发到 exchange1 判断路由符合，发送到队列queue1

![](https://image-static.segmentfault.com/411/027/4110271351-5db7b8925ca60_articlex)

在 RabbitMQ 的管理界面，当我们集群部署时可以看到 Nodes 节点中 Info 字段可能为 disc 也可能ram，表示了磁盘存储或内存储存。事实上，在集群部署的时候，我们至少要一个磁盘储存，它代表了将交换机，队列，绑定，用户等元数据持久化保存到磁盘，一遍重启 RabbitMQ 也能恢复到原先的状态，当只有一个节点时，必定是磁盘存储；而内存储存也有它的优势，就是效率更高速度更快

当报下列错误，表示你一定存在排他性队列，也就是设置了 exclusive 属性的队列，由于同一个连接创建的不同通道可以访问同一个队列，此时由于这个排他属性会得到资源被锁定错误，也就是下列的错误。
由此我们可以知道，若你把队列设置成了 exclusive 属性的，那么就别创建新的连接去访问同一个队列

ESOURCE_LOCKED - cannot obtain exclusive access to locked queue xxxxxx

欢迎关注公众号，一起学习进步

1. RabbitMQ 实战应用技巧

1.1. 前言

1.2. 概述

1.3. 交换机

1.3.1. Direct Exchage

1.3.2. Topic Exchange

1.3.3. Headers Exchage

1.3.4. Fanout Exchange

1.4. 复杂情况

1.5. springboot 配置

1.6. 队列属性

1.7. 交换机属性

1.8. 磁盘和内存

1.9. 报错案例

Just My Socks（注册教程内含优惠码）