共计 5284 个字符，预计需要花费 14 分钟才能阅读完成。
1. RabbitMQ 实战应用技巧
1.1. 前言
由于项目原因，之后会和 RabbitMQ 比较多的打交道，所以让我们来好好整理下 RabbitMQ 的应用实战技巧，尽量避免日后的采坑
1.2. 概述
RabbitMQ 有几个重要的概念：虚拟主机，交换机，队列和绑定
- 虚拟主机：一个虚拟主机持有一组交换机、队列和绑定，我们可以从虚拟主机层面的颗粒度进行权限控制
- 交换机：Exchange 用于转发消息，它并不存储消息，如果没有 Queue 队列绑定到 Exchange，它会直接丢弃掉生产者发来的数据。
交换机还有个关联的重要概念：路由键，消息转发到哪个队列根据路由键决定
- 绑定：就是绑定交换机和队列，它是 多对多的关系，也就是说多个交换机可以绑同一个队列，也可以一个交换机绑多个队列
1.3. 交换机
交换机有四种类型的模式Direct, topic, Headers and Fanout
1.3.1. Direct Exchage
Direct 模式使用的是 RabbitMQ 的默认交换机，也是最简单的模式，适合比较简单的场景
如下图所示，使用 Direct 模式，我们需要创建不同的队列，而默认交换机则通过 Routing key 路由键的值来决定转发到哪个队列，可以看到，路由键绑定队列是可以指定多个的
1.3.2. Topic Exchange
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!");1.3.3. Headers Exchage
Headers 也是根据规则匹配，但它不是根据路由键了，headers 有个自定义匹配规则，它将匹配键值设在了消息的 headers 属性上，当这些键值对有一对或者全部匹配时，消息才会被投递到对应队列，这种模式效率相对较低，一般不推荐使用
1.3.4. Fanout Exchange
Fanout 即为大名鼎鼎的广播模式了，它 不需要管路由键 ，会把消息发给绑定它的全部队列，就算配置了 路由键也会被忽略
1.4. 复杂情况
- 首先我们 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.#");
}
}1.5. springboot 配置
我们的常用配置如下
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);
}
}1.6. 队列属性
- 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 开销，但适合应对大量消息堆积的情况；因为当大量消息堆积时，内存也不够存放，会将消息转存到磁盘，这个过程也是比较耗时且过程中不能接收新的消息。如果需要将普通队列转换成惰性队列需要将原来的队列删除，重新创建个惰性队列绑定。
- 比如添加
1.7. 交换机属性
- 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
- 如下图：
1.8. 磁盘和内存
在 RabbitMQ 的管理界面，当我们集群部署时可以看到 Nodes 节点中 Info 字段可能为 disc 也可能ram，表示了磁盘存储或内存储存。事实上，在集群部署的时候，我们至少要一个磁盘储存，它代表了将交换机，队列，绑定，用户等元数据持久化保存到磁盘，一遍重启 RabbitMQ 也能恢复到原先的状态，当只有一个节点时，必定是磁盘存储；而内存储存也有它的优势，就是效率更高速度更快
1.9. 报错案例
- 当报下列错误，表示你一定存在排他性队列，也就是设置了
exclusive属性的队列，由于同一个连接创建的不同通道可以访问同一个队列，此时由于这个排他属性会得到资源被锁定错误，也就是下列的错误。 - 由此我们可以知道，若你把队列设置成了
exclusive属性的，那么就别创建新的连接去访问同一个队列
ESOURCE_LOCKED - cannot obtain exclusive access to locked queue xxxxxx
欢迎关注公众号，一起学习进步