根据我对Rabbit-mq的(有限的)经验,如果你为尚不存在的队列创建新的侦听器,则会自动创建该队列。我试图将Spring AMQP项目与rabbit-mq一起使用来设置侦听器,但是却出现错误。这是我的xml配置:
<rabbit:connection-factory id="rabbitConnectionFactory" host="172.16.45.1" username="test" password="password" /> <rabbit:listener-container connection-factory="rabbitConnectionFactory" > <rabbit:listener ref="testQueueListener" queue-names="test" /> </rabbit:listener-container> <bean id="testQueueListener" class="com.levelsbeyond.rabbit.TestQueueListener"> </bean>
我在RabbitMq日志中得到了这个:
=ERROR REPORT==== 3-May-2013::23:17:24 === connection <0.1652.0>, channel 1 - soft error: {amqp_error,not_found,"no queue 'test' in vhost '/'",'queue.declare'}
还有来自AMQP的类似错误:
2019-05-03 23:17:24,059 ERROR [org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer] (SimpleAsyncTaskExecutor-1) - Consumer received fatal exception on startup org.springframework.amqp.rabbit.listener.FatalListenerStartupException: Cannot prepare queue for listener. Either the queue doesn't exist or the broker will not allow us to use it.
从堆栈跟踪中可以看出,队列是在“被动”模式下创建的-谁能指出我将如何不使用被动模式来创建队列,所以我看不到此错误?还是我想念其他东西?
解决我问题的方法似乎是添加一个管理员。这是我的xml:
<rabbit:listener-container connection-factory="rabbitConnectionFactory" > <rabbit:listener ref="orderQueueListener" queues="test.order" /> </rabbit:listener-container> <rabbit:queue name="test.order"></rabbit:queue> <rabbit:admin id="amqpAdmin" connection-factory="rabbitConnectionFactory"/> <bean id="orderQueueListener" class="com.levelsbeyond.rabbit.OrderQueueListener"> </bean>
从带有Spring-Messaging和Spring-Amqp 1.4.5.RELEASE和Spring-Rabbit 1.4.5.RELEASE的Spring 4.2.x 开始,通过@Configuration类声明一些注释,就非常简单地声明了交换,队列和绑定:
@EnableRabbit @Configuration @PropertySources({ @PropertySource("classpath:rabbitMq.properties") }) public class RabbitMqConfig { private static final Logger logger = LoggerFactory.getLogger(RabbitMqConfig.class); @Value("${rabbitmq.host}") private String host; @Value("${rabbitmq.port:5672}") private int port; @Value("${rabbitmq.username}") private String username; @Value("${rabbitmq.password}") private String password; @Bean public ConnectionFactory connectionFactory() { CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host, port); connectionFactory.setUsername(username); connectionFactory.setPassword(password); logger.info("Creating connection factory with: " + username + "@" + host + ":" + port); return connectionFactory; } /** * Required for executing adminstration functions against an AMQP Broker */ @Bean public AmqpAdmin amqpAdmin() { return new RabbitAdmin(connectionFactory()); } /** * This queue will be declared. This means it will be created if it does not exist. Once declared, you can do something * like the following: * * @RabbitListener(queues = "#{@myDurableQueue}") * @Transactional * public void handleMyDurableQueueMessage(CustomDurableDto myMessage) { * // Anything you want! This can also return a non-void which will queue it back in to the queue attached to @RabbitListener * } */ @Bean public Queue myDurableQueue() { // This queue has the following properties: // name: my_durable // durable: true // exclusive: false // auto_delete: false return new Queue("my_durable", true, false, false); } /** * The following is a complete declaration of an exchange, a queue and a exchange-queue binding */ @Bean public TopicExchange emailExchange() { return new TopicExchange("email", true, false); } @Bean public Queue inboundEmailQueue() { return new Queue("email_inbound", true, false, false); } @Bean public Binding inboundEmailExchangeBinding() { // Important part is the routing key -- this is just an example return BindingBuilder.bind(inboundEmailQueue()).to(emailExchange()).with("from.*"); } }