在版本2.0之前的版本中，只有一种MessageListenerContainer—SimpleMessageListenerContainer; 2.0之后有第二个容器——DirectMessageListenerContainer...SimpleMessageListenerContainer 默认情况下，侦听器容器将启动单个使用者，该使用者将从队列接收消息。...提供了以下特性，但DirectMessageListenerContainer不提供: txSize—使用SimpleMessageListenerContainer，您可以将其设置为控制事务中传递的消息数量和...然而，与SimpleMessageListenerContainer相比，DirectMessageListenerContainer有以下优点: 在运行时添加和删除队列更有效;使用SimpleMessageListenerContainer...线程是跨使用者共享的，而不是为SimpleMessageListenerContainer中的每个使用者都有一个专用线程。但是，请参阅“线程和异步使用者”一节中有关连接工厂配置的重要说明。
如@RabbitListener、cloud-stream中StreamListener中底层实现都是基于该类，所以，理解SimpleMessageListenerContainer原理对理解spring...基本使用 1、SimpleMessageListenerContainer#addQueueNames()方法可以运行时添加监听队列，removeQueueNames()方法可以运行时移除监听队列； 2...@Bean public SimpleMessageListenerContainer messageListenerContainer(ConnectionFactory connectionFactory...){ SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); container.setConnectionFactory...类结构如下： 源码分析 方法入口 SimpleMessageListenerContainer类启动的入口是start()方法，该方法位于AbstractMessageListenerContainer
rabbitMQ精讲系列第二十一篇 高级整合应用第三篇 SimpleMessageListenerContainer 今天主要内容： SimpleMessageListenerContainer对象...SimpleMessageListenerContainer：简单消息监听容器。...代码示例： 在rabbitMQConfig类中添加SimpleMessageListenerContainer的bean.在simpleMessageListenerContainer可以设置很多东西。...connectionFactory){ SimpleMessageListenerContainer messageListenerContainer = new SimpleMessageListenerContainer...经过以上两个测试，我们的simpleMessageListenerContainer设置成功了。 相关代码以更新到git上了。 下节预告：MessageListenerAdapter
(SimpleMessageListenerContainer.java:921) at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access...$1600(SimpleMessageListenerContainer.java:83) at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer...$1600(SimpleMessageListenerContainer.java:83) at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer...$1600(SimpleMessageListenerContainer.java:83) at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer...$1600(SimpleMessageListenerContainer.java:83) at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer
$001(SimpleMessageListenerContainer.java:95) at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer...(SimpleMessageListenerContainer.java:1149) at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access...$1100(SimpleMessageListenerContainer.java:95) at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer...$001(SimpleMessageListenerContainer.java:95) at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer...$1100(SimpleMessageListenerContainer.java:95) at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer
(SimpleMessageListenerContainer.java:948) at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access...$1600(SimpleMessageListenerContainer.java:86) at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer...$1600(SimpleMessageListenerContainer.java:86) at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer...$1600(SimpleMessageListenerContainer.java:86) at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer...$1600(SimpleMessageListenerContainer.java:86) at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer
高效监听消息 SimpleMessageListenerContainer特征： ◆ 设置同时监听多个队列、自动启动、自动配置RabbitMQ ◆ 设置消费者数量(最大数量、最小数量、批量消费) ◆...simpleMessageListenerContainer = new SimpleMessageListenerContainer(connectionFactory); simpleMessageListenerContainer.setQueueNames...(3); // 最大线程数 simpleMessageListenerContainer.setMaxConcurrentConsumers(5); //...) { SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer.../*simpleMessageListenerContainer.setAcknowledgeMode(AcknowledgeMode.AUTO); simpleMessageListenerContainer.setMessageListener
org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer...MessageListenerAdapter listenerAdapter = new MessageListenerAdapter(messageListener, methodName); // 创建 SimpleMessageListenerContainer...SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); container.setConnectionFactory
2793af8966235f73254732ab255f91f.png RabbitMQ rabbitMQ 使用 messageConvert 解析 x-request-id，使用 SmartInitializingSingleton 获取 SimpleMessageListenerContainer...ae7890c53b8ef50525ebd403f33f347.png image.png @Slf4j @Component @ConditionalOnClass(SimpleMessageListenerContainer.class...class FromProcessorConfig implements SmartInitializingSingleton { @Autowired(required = false) SimpleMessageListenerContainer...listenerContainer; public void setAfterReceivePostProcessors(SimpleMessageListenerContainer listenerContainer...> obj = Class.forName("org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer");
注册，然后使用SimpleRabbitListenerContainerFactory和MethodRabbitListenerEndpoint进行属性配置，最终把创建的SimpleMessageListenerContainer...对象，然后使用SimpleRabbitListenerContainerFactory和MethodRabbitListenerEndpoint对SimpleMessageListenerContainer...进行配置，最终将SimpleMessageListenerContainer放入map容器中 RabbitListenerEndpointRegistry启动方法会遍历上一步的map容器，启动SimpleMessageListenerContainer...#run--->org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.AsyncMessageProcessingConsumer...// 3.向消息服务器发送ack return consumer.commitIfNecessary(isChannelLocallyTransacted()); } 4.5 小结 启动SimpleMessageListenerContainer
.*")); //设置监听 SimpleMessageListenerContainer container = new SimpleMessageListenerContainer...; 消费消息： 设置MessageListenerAdapter 这里面有几个类需要额外注意： RabbitTemplate: Spring实现的发送消息的模板，可以直接发送消息 SimpleMessageListenerContainer...container = new SimpleMessageListenerContainer(); container.setConnectionFactory(connectionFactory...container = new SimpleMessageListenerContainer(); container.setConnectionFactory(connectionFactory...container = new SimpleMessageListenerContainer(); container.setConnectionFactory(connectionFactory
注意：SimpleMessageListenerContainer 可以进行动态设置，比如在运行中的应用可以动态修改其消费者数量大小、接收消息的模式等。...messageListenerContainer(ConnectionFactory connectionFactory) { SimpleMessageListenerContainer...container = new SimpleMessageListenerContainer(connectionFactory); container.setQueues(queue001...messageListenerContainer(ConnectionFactory connectionFactory) { SimpleMessageListenerContainer...container = new SimpleMessageListenerContainer(connectionFactory); container.setQueues(queue001
~[spring-rabbit-2.1.2.RELEASE.jar:2.1.2.RELEASE] 11 at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer...$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1026) [spring-rabbit-2.1.2.RELEASE.jar...connect to: [192.168.110.133:5672] 30 2019-11-21 22:54:49.261 WARN 19532 --- [TaskExecutor-24] o.s.a.r.l.SimpleMessageListenerContainer...Connection refused: connect 31 2019-11-21 22:54:49.261 INFO 19532 --- [TaskExecutor-24] o.s.a.r.l.SimpleMessageListenerContainer...connect to: [192.168.110.133:5672] 33 2019-11-21 22:54:50.266 ERROR 19532 --- [TaskExecutor-25] o.s.a.r.l.SimpleMessageListenerContainer
生产者端需要配置RabbitAdmin 生产者端不像消费端的SimpleMessageListenerContainer一样，可以自动初始化RabbitAdmin对象，需要我们手动来创建。...####消费端核心配置SimpleMessageListenerContainer类 @Bean public SimpleMessageListenerContainer messageListenerContainer...() { SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); container.setConnectionFactory...return container; } ####SimpleMessageListenerContainer的doStart()方法 我们来看下SimpleMessageListenerContainer
(SimpleMessageListenerContainer.java:975) at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute...(SimpleMessageListenerContainer.java:921) at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access...$1600(SimpleMessageListenerContainer.java:83) at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer...$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1296) at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer...$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1202) at java.lang.Thread.run
org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1303) at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute...(SimpleMessageListenerContainer.java:817) at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute...(SimpleMessageListenerContainer.java:801) at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access...$700(SimpleMessageListenerContainer.java:77) at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer...$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1042) at java.lang.Thread.run
下面介绍一下两个实现SimpleMessageListenerContainer和DefaultMessageListenerContainer。 下面是类图（只包含了部分关键属性）： ?...SimpleMessageListenerContainer SimpleMessageListenerContainer使用MessageConsumer.setMessageListener()来监听消息...processMessage(message, session); } }); } DefaultMessageListenerContainer DefaultMessageListenerContainer和SimpleMessageListenerContainer...Spring Boot提供DefaultJmsListenerContainerFactoryConfigurer作为配置工具 SimpleJmsListenerContainerFactory，用来生产SimpleMessageListenerContainer
RabbitTemplate rabbitMq; /** * 手动批量创建队列的消费者 * @return */ @Bean public SimpleMessageListenerContainer..."+i); } final ConnectionFactory factory = rabbitMq.getConnectionFactory(); SimpleMessageListenerContainer...container = new SimpleMessageListenerContainer(factory); //加入所有的待消费队列到监听器内 container.setQueueNames
org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer...exchange) { return BindingBuilder.bind(queue).to(exchange).with(queueName); } @Bean SimpleMessageListenerContainer...connectionFactory, MessageListenerAdapter listenerAdapter) { SimpleMessageListenerContainer...container = new SimpleMessageListenerContainer(); container.setConnectionFactory(connectionFactory
步骤如下： 注入SimpleMessageListenerContainer 12345678910111213141516171819@Bean public SimpleMessageListenerContainer...simpleMessageListenerContainer(ConnectionFactory connectionFactory){ SimpleMessageListenerContainer...SimpleMessageListenerContainer中的两个属性可以完成设置，如下： concurrentConsumers：消费者的数量，默认1 maxConcurrentConsumers：...实现步骤： 设置消费者的确认模式为手动确认，使用的是SimpleMessageListenerContainer的API 12//设置消费者ack消息的模式，默认是自动，此处设置为手动...设置重回队列如下： SimpleMessageListenerContainer中设置默认的行为如下： 12//设置不重回队列，默认为true，即是消息被拒绝或者nack或者监听器抛出异常之后会重新返回队列
领取专属 10元无门槛券
手把手带您无忧上云
云服务器
ICP备案
对象存储
腾讯会议
实时音视频
