spring boot 集成RabbitMQ 接收消息及其高级特性
基本用法
连接和资源管理
管理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包含有以下参数：
@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:最新发布的消息被丢弃。
死信队列
消息变为死信的情况：
- 消息被拒绝
- 消息过期
- 队列长度达到最大长度
如何设置死信队列
创建队列时设置队列的死信路由键即可。订阅此路由的队列就可以收到死信消息。
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
