SpringAMQP 消息容器 - SimpleMessageListenerContainer

133 阅读 0 评论 88 点赞

我是靠谱客的博主寒冷大神，这篇文章主要介绍SpringAMQP 消息容器 - SimpleMessageListenerContainer，现在分享给大家，希望可以做个参考。

SimpleMessageListenerContainer 即简单消息监听容器。

这个类非常的强大，我们可以对他进行很多的设置，用对于消费者的配置项，这个类都可以满足。它有监听单个或多个队列、自动启动、自动声明功能。
它可以设置事务特性、事务管理器、事务属性、事务并发、是否开启事务、回滚消息等。但是我们在实际生产中，很少使用事务，基本都是采用补偿机制。
它可以设置消费者数量、最小最大数量、批量消费。
它可以设置消息确认和自动确认模式、是否重回队列、异常捕获 Handler 函数。
它可以设置消费者标签生成策略、是否独占模式、消费者属性等。
它还可以设置具体的监听器、消息转换器等等。

注意: SimpleMessageListenerContainer 可以进行动态设置，比如在运行中的应用可以动态的修改其消费者数量的大小、接收消息的模式等。

很多基于 rabbitMQ 的自制定化后端管控台在进行设置的时候，也是根据这一去实现的。所以可以看出 SpringAMQP 非常的强大。

代码示例：

复制代码

1
2
3
4
代码地址：
https://github.com/hmilyos/rabbitmqdemo.git
rabbitmq-api 项目下
复制代码

在上一节的 SpringAMQP 之 RabbitTemplate 的 RabbitMQConfig 原有代码的基础上加上

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

@Bean
//connectionFactory 也是要和最上面方法名保持一致
public SimpleMessageListenerContainer messageContainer(ConnectionFactory connectionFactory) {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
container.setQueues(queue001(), queue002(), queue003());
//监听的队列
container.setConcurrentConsumers(1);
//当前的消费者数量
container.setMaxConcurrentConsumers(5); //
最大的消费者数量
container.setDefaultRequeueRejected(false); //是否重回队列
container.setAcknowledgeMode(AcknowledgeMode.AUTO); //签收模式
container.setExposeListenerChannel(true);
container.setConsumerTagStrategy(new ConsumerTagStrategy() {
//消费端的标签策略
@Override
public String createConsumerTag(String queue) {
return queue + "_" + UUID.randomUUID().toString();
}
});
container.setMessageListener(new ChannelAwareMessageListener() {
@Override
public void onMessage(Message message, Channel channel) throws Exception {
String msg = new String(message.getBody());
log.info("----------消费者: " + msg);
}
});
return container;
}
复制代码

单元测试里面还是继续使用上一次的代码

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

@Test
public void testSendMessage2() throws Exception {
//1 创建消息
MessageProperties messageProperties = new MessageProperties();
messageProperties.setContentType("text/plain");
Message message = new Message("mq 消息1234
--spring.abc".getBytes(), messageProperties);
rabbitTemplate.send("topic001", "spring.abc", message);
rabbitTemplate.convertAndSend("topic001", "spring.amqp", "hello object message send!
-spring.amqp");
rabbitTemplate.convertAndSend("topic002", "rabbit.abc", "hello object message send!
-rabbit.abc");
}
复制代码