spring boot 集成RabbitMQ 接收消息及其高级特性

原创
11/20 18:21
阅读数 3.1K

基本用法

连接和资源管理

管理MQ的核心类:ConnectionFactory

ConnectionFactory封装了com.rabbitmq.client.Connection,官方提供的实现类为CachingConnectionFactory,它缓存了创建的信道(Channel),默认为25个,可以通过setChannelCacheSize来更改数量。

  @Bean(name="cwagConnectionFactory")
    public ConnectionFactory cwagConnectionFactory(){
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost(host);
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        return connectionFactory;
    }

@RabbitListener的那些配置

从1.4版本开始,@RabbitListener(queues = "myQueue")可以接收来自myQueue队列的消息,但是这个队列必须在路由里面存在。
从1.5版本开始,queue可以自动声明和绑定。

  • 一个配置的列子:
 @RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "myQueue", durable = "true"),
        exchange = @Exchange(value = "auto.exch", ignoreDeclarationExceptions = "true"),
        key = "orderRoutingKey")
  )
  public void processOrder(String data) {
    ...
  }

其中,@QueueBinding包含有以下参数:

  • value:类型为@Queue,如果只配置为@Queue,没有配置value属性,则会创建一个匿名队列(exclusive, auto-delete)例如:
 @RabbitListener(bindings = @QueueBinding(
        value = @Queue,
        exchange = @Exchange(value = "auto.exch", ignoreDeclarationExceptions = "true"))
  )
  • exchange:类型为@Exchange,Exchange支持的类型为DIRECT, FANOUT, TOPIC and HEADERS,例如:
 exchange = @Exchange(value = "auto.exch",type = ExchangeTypes.TOPIC,autoDelete = "tue",durable = "true"),
可能遇到的问题:
如何动态设置Qeueu的名称?

@Queue支持properties或者SpEL表达式定义:

通过property定义:${...}

 @RabbitListener(queues = { "${queue1}", "${queue2}" } )
 @RabbitListener(queues = "#{'${property.with.comma.delimited.queue.names}'.split(',')}" )

通过SpEL表达式定义:#{...}

 @RabbitListener(queues = "#{queueName}" )
 ....
 
 //在Bean中声明
 @Bean
   public String queueName(){
       return "helloQueue";
   }
 
如何通过Exchange,RoutingKey进行连接绑定?
 @RabbitListener(bindings = @QueueBinding(
        value = @Queue,//设置一个匿名队列
        exchange = @Exchange(value = "auto.exch"),key="routingKey")
  )

自己手动配置MQ连接,例如:系统中要连接多个MQ地址;需要配置多个线程处理消息

  • ConnectionFactory:配置MQ连接
  @Bean(name="cwagConnectionFactory")
    public ConnectionFactory cwagConnectionFactory(){
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost(host);
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        return connectionFactory;
    }
  • RabbitListenerContainerFactory 构造ListenerContainerFactory
    @Bean(name="myListenerFactory")
    public SimpleRabbitListenerContainerFactory firstFactory(
            SimpleRabbitListenerContainerFactoryConfigurer configurer,
            @Qualifier("myConnectionFactory") ConnectionFactory connectionFactory
         ) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConcurrentConsumers(3);//默认开启3个线程的线程池
        factory.setMaxConcurrentConsumers(10);//最大线程为10个
        configurer.configure(factory, connectionFactory);
        return factory;
    }
  • 引用构造ListenerContainerFactory:
  @RabbitHandler
  @RabbitListener(queues = "MyQueue",containerFactory = "myListenerFactory")
    public void receiveApply(@Payload byte[] msg) throws IOException {
        log.info("received from MyQueue"  );
    }

高级特性

队列长度

  • 如何设置队列长度:设置最大消息数量(x-max-length)或者最大容量(x-max-length-bytes)
        Map<String,Object> map = new HashMap<>();
        //设置最大长度
        map.put("x-max-length",5);
        return new Queue("5-length",false,false,false,map);
  • 投递消息数量超多队列长度
    • 默认情况下(overflow:drop-head),当消息数量超多最大容量后,队列头部(最老的消息)消息会被丢弃或者变为死信(死信路由存在的时候才会发到对应的路由)。可以通过overflow来设置默认规则。
  • overflow的规则
    • drop-head
    • reject-publish:最新发布的消息被丢弃。

死信队列

消息变为死信的情况:
  1. 消息被拒绝
  2. 消息过期
  3. 队列长度达到最大长度
如何设置死信队列

创建队列时设置队列的死信路由键即可。订阅此路由的队列就可以收到死信消息。

        Map<String,Object> map = new HashMap<>();
        //设置最大长度
        map.put("x-max-length",5);
        //设置死信路由
        map.put("x-dead-letter-exchange","my-dead-exchange");
        map.put("x-dead-letter-routing-key","dead-r-5");
        return new Queue("5-length",false,false,false,map);

延迟队列

延迟队列就是某个时间段过后消息被投递到此队列。可以利用消息超时后被进入到死信队列这个特点来实现。 在这里插入图片描述 如上图所示,绑定了x-dead-letter-exchage,x-dead-letter-routing-key的队列就是延迟队列。

Tips:

对于使用过程中存在的问题,仔细阅读官方文档是最有效的方法, 参考文档:https://docs.spring.io/spring-amqp/reference/html/_reference.html#broker-configuration 注解配置:https://docs.spring.io/spring-amqp/docs/1.7.11.RELEASE/reference/html/_reference.html#async-annotation-driven

展开阅读全文
打赏
2
6 收藏
分享
加载中
更多评论
打赏
0 评论
6 收藏
2
分享
OSCHINA
登录后可查看更多优质内容
返回顶部
顶部