Java 如何让 Spring RabbitMQ 创建一个新的队列？

How to get Spring RabbitMQ to create a new Queue?

提问by eric

In my (limited) experience with rabbit-mq, if you create a new listener for a queue that doesn't exist yet, the queue is automatically created. I'm trying to use the Spring AMQP project with rabbit-mq to set up a listener, and I'm getting an error instead. This is my xml config:

根据我对 rabbit-mq 的（有限）经验，如果您为尚不存在的队列创建新侦听器，则会自动创建该队列。我正在尝试使用带有 rabbit-mq 的 Spring AMQP 项目来设置侦听器，但出现错误。这是我的 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>

I get this in my RabbitMq logs:

我在我的 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'}

And a similar error from AMQP:

来自 AMQP 的类似错误：

2013-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.

It would seem from the stack trace that the queue is getting created in a "passive" mode- Can anyone point out how I would create the queue not using the passive mode so I don't see this error? Or am I missing something else?

从堆栈跟踪看来，队列是在“被动”模式下创建的 - 谁能指出我将如何不使用被动模式创建队列，所以我没有看到这个错误？还是我错过了其他东西？

采纳答案by eric

What seemed to resolve my issue was adding an admin. Here is my xml:

似乎解决我的问题的是添加管理员。这是我的 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>

回答by Homer6

Can you add this after your connection tag, but before the listener:

您可以在连接标记之后但在侦听器之前添加此内容吗：

<rabbit:queue name="test" auto-delete="true" durable="false" passive="false" />

Unfortunately, according to the XSD schema, the passive attribute (listed above) is not valid. However, in every queue_declare implementation I've seen, passive has been a valid queue_declare parameter. I'm curious to see whether that will work or whether they plan to support it in future.

不幸的是，根据 XSD 架构，被动属性（如上所列）无效。然而，在我见过的每一个 queue_declare 实现中，passive 都是一个有效的 queue_declare 参数。我很好奇这是否会奏效，或者他们是否计划在未来支持它。

Here is the full list of options for a queue declaration: http://www.rabbitmq.com/amqp-0-9-1-reference.html#class.queue

以下是队列声明的完整选项列表：http: //www.rabbitmq.com/amqp-0-9-1-reference.html#class.queue

And here is the full XSD for the spring rabbit schema (with comments included): http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd

这是 spring rabbit 模式的完整 XSD（包括注释）：http: //www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd

回答by Jaymes Bearden

Older thread, but this still shows up pretty high on Google, so here's some newer information:

较旧的线程，但这在 Google 上的显示仍然很高，所以这里有一些更新的信息：

2015-11-23

2015-11-23

Since Spring 4.2.xwith Spring-Messaging and Spring-Amqp 1.4.5.RELEASEand Spring-Rabbit 1.4.5.RELEASE, declaring exchanges, queues and bindings has become very simple through an @Configuration class some annotations:

从带有 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.*");
    }
}

Some sources and documentation to help:

一些可提供帮助的来源和文档：

1. Spring annotations
2. Declaring/configuration RabbitMQ for queue/binding support
3. Direct exchange binding (for when routing key doesn't matter)

1. Spring注解
2. 声明/配置 RabbitMQ 以支持队列/绑定
3. 直接交换绑定（当路由密钥无关紧要时）

Note: Looks like I missed a version -- starting with Spring AMQP 1.5, things get even easier as you can declare the full binding right at the listener!

注意：看起来我错过了一个版本——从Spring AMQP 1.5 开始，事情变得更加容易，因为您可以在侦听器上声明完整的绑定！

回答by Rafael Renan Pacheco

As of Spring Boot 2.1.6and Spring AMQP 2.1.7you can create queues during startup if they don't exists with this:

从Spring Boot 2.1.6和Spring AMQP 2.1.7 开始，如果队列不存在，您可以在启动期间创建队列：

@Component
public class QueueConfig {

    private AmqpAdmin amqpAdmin;

    public QueueConfig(AmqpAdmin amqpAdmin) {
        this.amqpAdmin = amqpAdmin;
    }

    @PostConstruct
    public void createQueues() {
        amqpAdmin.declareQueue(new Queue("queue_one", true));
        amqpAdmin.declareQueue(new Queue("queue_two", true));
    }
}