RabbitMQ实战应用技巧

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. 复杂状况

1. 首先咱们Direct模式，一个生产者一个消费者的状况，也就对应了一个发送者和一个队列A接收，这是没有疑问的，发送什么接收什么
2. 当Direct模式，一个生产者发消息，开启多个消费者也就是多个相同queue，此时消息由多个消费者均匀分摊，不会重复消费（前提ack正常）
3. 当Topic模式，一个交换机绑定两个队列，路由键有重叠关系，以下代码，此时指定路由键topic.message发送消息，队列queueMessage和queueMessages都能接收到相同消息，也就是说，topic模式能够实现相似于广播模式的形式，甚至更加灵活，它可否转发到消息由路由键决定。
4. 相比于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. 队列属性

1. queue ： 队列的名字
2. durable ： 为true表示队列中数据持久化到磁盘，能够防止mq宕机重启数据丢失
3. exclusive ： 为true表示排他性，只容许一个当前链接访问该队列，当前已链接就不容许新的链接进入不然报错，当链接断开当前队列会销毁
4. autoDelete ： 为true表示自动删除，当没有Connection链接到队列的时候，会自动删除
5. 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. 交换机属性

1. exchange ： 交换机名称
2. type ： 交换机类型
3. durable ： 持久化，同队列
4. autoDelete ： 是否自动删除，同队列
5. internal ： 若为true，表示这个exchange不能够被client用来推送消息，仅用来进行exchange和exchange之间的绑定。
6. 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

今日教学视频：RabbitMQ消息队列从入门到精通，长按图片到百度云

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