配置 Broker

Doubao 中英对照 Configuring the Broker

AMQP 规范描述了如何使用该协议在 broker 上配置队列、交换机和绑定。这些操作（从 0.8 规范起可移植）存在于 org.springframework.amqp.core 包的 AmqpAdmin 接口中。该类的 RabbitMQ 实现是位于 org.springframework.amqp.rabbit.core 包中的 RabbitAdmin。

AmqpAdmin 接口基于使用 Spring AMQP 领域抽象，如下清单所示：

public interface AmqpAdmin {

    // Exchange Operations

    void declareExchange(Exchange exchange);

    void deleteExchange(String exchangeName);

    // Queue Operations

    Queue declareQueue();

    String declareQueue(Queue queue);

    boolean deleteQueue(String queueName);

    void deleteQueue(String queueName, boolean unused, boolean empty);

    void purgeQueue(String queueName, boolean noWait);

    // Binding Operations

    void declareBinding(Binding binding);

    void removeBinding(Binding binding);

    Properties getQueueProperties(String queueName);

    QueueInformation getQueueInfo(String queueName);

}

另请参见作用域操作。

getQueueProperties() 方法返回关于队列的一些有限信息（消息数和消费者数）。返回属性的键在 RabbitAdmin 中以常量形式提供（QUEUE_NAME、QUEUE_MESSAGE_COUNT 和 QUEUE_CONSUMER_COUNT）。getQueueInfo() 返回一个便利的 QueueInformation 数据对象。

无参的 declareQueue() 方法在 broker 上定义一个名称自动生成的队列。这个自动生成队列的附加属性为 exclusive=true、autoDelete=true 和 durable=false。

declareQueue(Queue queue) 方法接收一个 Queue 对象并返回声明队列的名称。如果所提供 Queue 的 name 属性是空 String，broker 就用生成的名称声明队列。该名称返回给调用者。该名称还被添加到 Queue 的 actualName 属性。你只能通过直接调用 RabbitAdmin 以编程方式使用此功能。在应用上下文中以声明方式定义队列时使用 admin 的自动声明，你可以把 name 属性设置为 ""（空字符串）。然后由 broker 创建名称。从 2.1 版本开始，监听器容器可以使用这种类型的队列。更多信息参见容器与 Broker 命名队列。

这与 AnonymousQueue 不同，在 AnonymousQueue 中框架生成唯一的（UUID）名称，并把 durable 设为 false、exclusive 和 autoDelete 设为 true。name 属性为空（或缺失）的 <rabbit:queue/> 总是创建一个 AnonymousQueue。

参见 AnonymousQueue，了解为什么 AnonymousQueue 优先于 broker 生成的队列名，以及如何控制名称格式。从 2.1 版本开始，匿名队列声明时默认把参数 Queue.X_QUEUE_LEADER_LOCATOR 设置为 client-local。这确保队列声明在应用所连接的节点上。声明式队列必须有固定名称，因为它们可能在上下文中其他地方被引用——例如下面示例中所示的监听器：

<rabbit:listener-container>
    <rabbit:listener ref="listener" queue-names="#{someQueue.name}" />
</rabbit:listener-container>

参见交换机、队列和绑定的自动声明。

此接口的 RabbitMQ 实现是 RabbitAdmin，使用 Spring XML 配置时如下例所示：

<rabbit:connection-factory id="connectionFactory"/>

<rabbit:admin id="amqpAdmin" connection-factory="connectionFactory"/>

当 CachingConnectionFactory 缓存模式为 CHANNEL（默认）时，RabbitAdmin 实现会自动懒式声明在同一 ApplicationContext 中声明的队列、交换机和绑定。这些组件在向 broker 打开一个 Connection 时即被声明。有一些命名空间特性让这非常方便——例如在 Stocks 示例应用中，我们有如下内容：

<rabbit:queue id="tradeQueue"/>

<rabbit:queue id="marketDataQueue"/>

<fanout-exchange name="broadcast.responses"
                 xmlns="http://www.springframework.org/schema/rabbit">
    <bindings>
        <binding queue="tradeQueue"/>
    </bindings>
</fanout-exchange>

<topic-exchange name="app.stock.marketdata"
                xmlns="http://www.springframework.org/schema/rabbit">
    <bindings>
        <binding queue="marketDataQueue" pattern="${stocks.quote.pattern}"/>
    </bindings>
</topic-exchange>

在上面的示例中，我们使用匿名队列（实际上，内部就是由框架而非 broker 生成名称的队列），并按 ID 引用它们。我们也可以声明带显式名称的队列，这些名称同时作为上下文中其 bean 定义的标识符。下例配置了一个带显式名称的队列：

<rabbit:queue name="stocks.trade.queue"/>

提示

你可以同时提供 id 和 name 属性。这让你可以用独立于队列名的 ID 引用队列（例如在绑定中）。它还允许使用标准 Spring 特性（例如队列名的属性占位符和 SpEL 表达式）。当你把 name 用作 bean 标识符时，这些特性不可用。

队列可以用附加参数配置——例如 x-message-ttl。使用命名空间支持时，它们以参数名/参数值对的 Map 形式提供，使用 <rabbit:queue-arguments> 元素定义。下例展示了如何这样做：

<rabbit:queue name="withArguments">
    <rabbit:queue-arguments>
        <entry key="x-dead-letter-exchange" value="myDLX"/>
        <entry key="x-dead-letter-routing-key" value="dlqRK"/>
    </rabbit:queue-arguments>
</rabbit:queue>

默认情况下，参数假定为字符串类型。对于其他类型的参数，你必须提供类型。下例展示了如何指定类型：

<rabbit:queue name="withArguments">
    <rabbit:queue-arguments value-type="java.lang.Long">
        <entry key="x-message-ttl" value="100"/>
    </rabbit:queue-arguments>
</rabbit:queue>

提供混合类型的参数时，你必须为每个 entry 元素提供类型。下例展示了如何这样做：

<rabbit:queue name="withArguments">
    <rabbit:queue-arguments>
        <entry key="x-message-ttl">
            <value type="java.lang.Long">100</value>
        </entry>
        <entry key="x-dead-letter-exchange" value="myDLX"/>
        <entry key="x-dead-letter-routing-key" value="dlqRK"/>
    </rabbit:queue-arguments>
</rabbit:queue>

对于 Spring Framework 3.2 及更高版本，可以更简洁地声明，如下所示：

<rabbit:queue name="withArguments">
    <rabbit:queue-arguments>
        <entry key="x-message-ttl" value="100" value-type="java.lang.Long"/>
        <entry key="x-ha-policy" value="all"/>
    </rabbit:queue-arguments>
</rabbit:queue>

使用 Java 配置时，Queue.X_QUEUE_LEADER_LOCATOR 参数作为一等属性通过 Queue 类上的 setLeaderLocator() 方法支持。从 2.1 版本开始，匿名队列声明时默认把此属性设置为 client-local。这确保队列声明在应用所连接的节点上。

important

RabbitMQ broker 不允许声明参数不匹配的队列。例如，如果一个 queue 已存在且没有 time to live 参数，而你尝试用（例如）key="x-message-ttl" value="100" 声明它，就会抛出异常。

默认情况下，发生任何异常时 RabbitAdmin 立即停止处理所有声明。这可能导致下游问题，例如监听器容器初始化失败，因为另一个队列（在出错队列之后定义的）没有被声明。

可以通过把 RabbitAdmin 实例上的 ignore-declaration-exceptions 属性设置为 true 来修改此行为。此选项指示 RabbitAdmin 记录异常并继续声明其他元素。使用 Java 配置 RabbitAdmin 时，此属性名为 ignoreDeclarationExceptions。这是应用于所有元素的全局设置。队列、交换机和绑定有一个类似的属性，仅适用于这些元素。

在 1.6 版本之前，此属性仅在 channel 上发生 IOException 时生效，例如当前属性与期望属性不匹配时。现在，此属性对任何异常都生效，包括 TimeoutException 等。

此外，任何声明异常都会导致发布一个 DeclarationExceptionEvent，它是一个 ApplicationEvent，可由上下文中的任何 ApplicationListener 消费。该事件包含对 admin、正被声明的元素和 Throwable 的引用。

Headers 交换机

从 1.3 版本开始，你可以配置 HeadersExchange 按多个头匹配。你还可以指定是必须匹配任意头还是全部头。下例展示了如何这样做：

<rabbit:headers-exchange name="headers-test">
    <rabbit:bindings>
        <rabbit:binding queue="bucket">
            <rabbit:binding-arguments>
                <entry key="foo" value="bar"/>
                <entry key="baz" value="qux"/>
                <entry key="x-match" value="all"/>
            </rabbit:binding-arguments>
        </rabbit:binding>
    </rabbit:bindings>
</rabbit:headers-exchange>

从 1.6 版本开始，你可以用 internal 标志配置 Exchanges（默认为 false），这样的 Exchange 会通过 RabbitAdmin 在 Broker 上正确配置（如果应用上下文中存在）。如果交换机的 internal 标志为 true，RabbitMQ 不让客户端使用该交换机。这对死信交换机或交换机到交换机绑定很有用——你不希望交换机被发布者直接使用。

要了解如何用 Java 配置 AMQP 基础设施，请看 Stock 示例应用，其中有 @Configuration 类 AbstractStockRabbitConfiguration，它又有 RabbitClientConfiguration 和 RabbitServerConfiguration 子类。下面的清单展示了 AbstractStockRabbitConfiguration 的代码：

@Configuration
public abstract class AbstractStockAppRabbitConfiguration {

    @Bean
    public CachingConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory =
            new CachingConnectionFactory("localhost");
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        return connectionFactory;
    }

    @Bean
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate template = new RabbitTemplate(connectionFactory());
        template.setMessageConverter(jsonMessageConverter());
        configureRabbitTemplate(template);
        return template;
    }

    @Bean
    public Jackson2JsonMessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public TopicExchange marketDataExchange() {
        return new TopicExchange("app.stock.marketdata");
    }

    // additional code omitted for brevity

}

在 Stock 应用中，服务器使用下面的 @Configuration 类配置：

@Configuration
public class RabbitServerConfiguration extends AbstractStockAppRabbitConfiguration  {

    @Bean
    public Queue stockRequestQueue() {
        return new Queue("app.stock.request");
    }
}

这就是 @Configuration 类整个继承链的终点。最终结果是 TopicExchange 和 Queue 在应用启动时声明到 broker。服务器配置中没有 TopicExchange 到队列的绑定，因为那是在客户端应用中完成的。不过，股票请求队列会自动绑定到 AMQP 默认交换机。此行为由规范定义。

客户端 @Configuration 类更有意思一些。它的声明如下：

@Configuration
public class RabbitClientConfiguration extends AbstractStockAppRabbitConfiguration {

    @Value("${stocks.quote.pattern}")
    private String marketDataRoutingKey;

    @Bean
    public Queue marketDataQueue() {
        return amqpAdmin().declareQueue();
    }

    /**
     * Binds to the market data exchange.
     * Interested in any stock quotes
     * that match its routing key.
     */
    @Bean
    public Binding marketDataBinding() {
        return BindingBuilder.bind(
                marketDataQueue()).to(marketDataExchange()).with(marketDataRoutingKey);
    }

    // additional code omitted for brevity

}

客户端通过 AmqpAdmin 上的 declareQueue() 方法声明另一个队列。它用一个外部化在属性文件中的路由模式把该队列绑定到行情交换机。

队列和交换机的 Builder API

1.6 版本引入了一个便捷的流式 API，用于在使用 Java 配置时配置 Queue 和 Exchange 对象。下例展示了如何使用它：

@Bean
public Queue queue() {
    return QueueBuilder.nonDurable("foo")
        .autoDelete()
        .exclusive()
        .withArgument("foo", "bar")
        .build();
}

@Bean
public Exchange exchange() {
  return ExchangeBuilder.directExchange("foo")
      .autoDelete()
      .internal()
      .withArgument("foo", "bar")
      .build();
}

更多信息参见 org.springframework.amqp.core.QueueBuilder 和 org.springframework.amqp.core.ExchangeBuilder 的 Javadoc。

从 2.0 版本开始，ExchangeBuilder 现在默认创建持久化交换机，以与各个 AbstractExchange 类上的简单构造函数保持一致。要用 builder 创建非持久化交换机，在调用 .build() 之前使用 .durable(false)。不再提供无参数的 durable() 方法。

2.2 版本引入了流式 API 来添加“众所周知的”交换机和队列参数…​

@Bean
public Queue allArgs1() {
    return QueueBuilder.nonDurable("all.args.1")
            .ttl(1000)
            .expires(200_000)
            .maxLength(42)
            .maxLengthBytes(10_000)
            .overflow(Overflow.rejectPublish)
            .deadLetterExchange("dlx")
            .deadLetterRoutingKey("dlrk")
            .maxPriority(4)
            .lazy()
            .leaderLocator(LeaderLocator.minLeaders)
            .singleActiveConsumer()
            .build();
}

@Bean
public DirectExchange ex() {
    return ExchangeBuilder.directExchange("ex.with.alternate")
            .durable(true)
            .alternate("alternate")
            .build();
}

声明交换机、队列和绑定的集合

你可以把 Declarable 对象（Queue、Exchange 和 Binding）的集合包装在 Declarables 对象中。RabbitAdmin 在应用上下文中检测此类 bean（以及离散的 Declarable bean），并在建立连接时（初始时和连接失败后）在 broker 上声明所包含的对象。下例展示了如何这样做：

@Configuration
public static class Config {

    @Bean
    public CachingConnectionFactory cf() {
        return new CachingConnectionFactory("localhost");
    }

    @Bean
    public RabbitAdmin admin(ConnectionFactory cf) {
        return new RabbitAdmin(cf);
    }

    @Bean
    public DirectExchange e1() {
        return new DirectExchange("e1", false, true);
    }

    @Bean
    public Queue q1() {
        return new Queue("q1", false, false, true);
    }

    @Bean
    public Binding b1() {
        return BindingBuilder.bind(q1()).to(e1()).with("k1");
    }

    @Bean
    public Declarables es() {
        return new Declarables(
                new DirectExchange("e2", false, true),
                new DirectExchange("e3", false, true));
    }

    @Bean
    public Declarables qs() {
        return new Declarables(
                new Queue("q2", false, false, true),
                new Queue("q3", false, false, true));
    }

    @Bean
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public Declarables prototypes() {
        return new Declarables(new Queue(this.prototypeQueueName, false, false, true));
    }

    @Bean
    public Declarables bs() {
        return new Declarables(
                new Binding("q2", DestinationType.QUEUE, "e2", "k2", null),
                new Binding("q3", DestinationType.QUEUE, "e3", "k3", null));
    }

    @Bean
    public Declarables ds() {
        return new Declarables(
                new DirectExchange("e4", false, true),
                new Queue("q4", false, false, true),
                new Binding("q4", DestinationType.QUEUE, "e4", "k4", null));
    }

}

important

在 2.1 之前的版本中，你可以通过定义 Collection<Declarable> 类型的 bean 来声明多个 Declarable 实例。这在某些情况下会导致不良副作用，因为 admin 必须遍历所有 Collection<?> bean。

2.2 版本向 Declarables 添加了 getDeclarablesByType 方法；这可以作为便利方法使用，例如在声明监听器容器 bean 时。

public SimpleMessageListenerContainer container(ConnectionFactory connectionFactory,
        Declarables mixedDeclarables, MessageListener listener) {

    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
    container.setQueues(mixedDeclarables.getDeclarablesByType(Queue.class).toArray(new Queue[0]));
    container.setMessageListener(listener);
    return container;
}

条件声明

默认情况下，应用上下文中所有 RabbitAdmin 实例（假设它们的 auto-startup="true"）都会声明所有队列、交换机和绑定。

从 2.1.9 版本开始，RabbitAdmin 有一个新属性 explicitDeclarationsOnly（默认为 false）；设置为 true 时，admin 只声明明确配置为由该 admin 声明的 bean。

备注

从 1.2 版本开始，你可以有条件地声明这些元素。当一个应用连接多个 broker、需要指定某个元素应在哪些 broker 上声明时，这特别有用。

表示这些元素的类实现了 Declarable，它有两个方法：shouldDeclare() 和 getDeclaringAdmins()。RabbitAdmin 使用这些方法确定某个实例是否实际应在其 Connection 上处理声明。

这些属性在命名空间中作为属性可用，如下例所示：

<rabbit:admin id="admin1" connection-factory="CF1" />

<rabbit:admin id="admin2" connection-factory="CF2" />

<rabbit:admin id="admin3" connection-factory="CF3" explicit-declarations-only="true" />

<rabbit:queue id="declaredByAdmin1AndAdmin2Implicitly" />

<rabbit:queue id="declaredByAdmin1AndAdmin2" declared-by="admin1, admin2" />

<rabbit:queue id="declaredByAdmin1Only" declared-by="admin1" />

<rabbit:queue id="notDeclaredByAllExceptAdmin3" auto-declare="false" />

<rabbit:direct-exchange name="direct" declared-by="admin1, admin2">
    <rabbit:bindings>
        <rabbit:binding key="foo" queue="bar"/>
    </rabbit:bindings>
</rabbit:direct-exchange>

备注

默认情况下，auto-declare 属性为 true，如果未提供 declared-by（或为空），则所有 RabbitAdmin 实例都声明该对象（只要 admin 的 auto-startup 属性为 true（默认），且 admin 的 explicit-declarations-only 属性为 false）。

类似地，你可以使用基于 Java 的 @Configuration 达到同样效果。下例中，组件由 admin1 声明，但不由 admin2 声明：

@Bean
public RabbitAdmin admin1() {
    return new RabbitAdmin(cf1());
}

@Bean
public RabbitAdmin admin2() {
    return new RabbitAdmin(cf2());
}

@Bean
public Queue queue() {
    Queue queue = new Queue("foo");
    queue.setAdminsThatShouldDeclare(admin1());
    return queue;
}

@Bean
public Exchange exchange() {
    DirectExchange exchange = new DirectExchange("bar");
    exchange.setAdminsThatShouldDeclare(admin1());
    return exchange;
}

@Bean
public Binding binding() {
    Binding binding = new Binding("foo", DestinationType.QUEUE, exchange().getName(), "foo", null);
    binding.setAdminsThatShouldDeclare(admin1());
    return binding;
}

关于 id 和 name 属性的说明

<rabbit:queue/> 和 <rabbit:exchange/> 元素上的 name 属性反映 broker 中实体的名称。对于队列，如果省略 name，会创建一个匿名队列（参见 AnonymousQueue）。

在 2.0 之前的版本中，name 还会注册为 bean 名称别名（类似于 <bean/> 元素上的 name）。

这导致了两个问题：

• 它阻止了用相同名称声明队列和交换机。
• 如果别名包含 SpEL 表达式（#{…​}），别名无法解析。

从 2.0 版本开始，如果你同时用 id 和 name 属性声明这些元素之一，name 不再声明为 bean 名称别名。如果你希望用相同的 name 声明队列和交换机，必须提供 id。

如果元素只有 name 属性，则没有变化。仍然可以用 name 引用该 bean——例如在绑定声明中。不过，如果名称包含 SpEL，你仍然无法引用它——必须提供 id 用于引用。

AnonymousQueue

一般来说，当你需要一个唯一命名、独占、自动删除的队列时，我们建议使用 AnonymousQueue，而不是 broker 定义的队列名（使用 "" 作为 Queue 名会导致 broker 生成队列名）。

这是因为：

1. 队列实际上是在与 broker 建立连接时才声明的。这是在 bean 创建并装配很久之后。使用队列的 bean 需要知道它的名称。事实上，应用启动时 broker 甚至可能没有运行。
2. 如果与 broker 的连接因某种原因丢失，admin 会用相同名称重新声明 AnonymousQueue。如果我们使用 broker 声明的队列，队列名就会改变。

你可以控制 AnonymousQueue 实例使用的队列名格式。

默认情况下，队列名以 spring.gen- 为前缀，后跟 UUID 的 base64 表示——例如：spring.gen-MRBv9sqISkuCiPfOYfpo4g。

你可以在构造参数中提供 AnonymousQueue.NamingStrategy 实现。下例展示了如何这样做：

@Bean
public Queue anon1() {
    return new AnonymousQueue();
}

@Bean
public Queue anon2() {
    return new AnonymousQueue(new AnonymousQueue.Base64UrlNamingStrategy("something-"));
}

@Bean
public Queue anon3() {
    return new AnonymousQueue(AnonymousQueue.UUIDNamingStrategy.DEFAULT);
}

第一个 bean 生成的队列名以 spring.gen- 为前缀，后跟 UUID 的 base64 表示——例如：spring.gen-MRBv9sqISkuCiPfOYfpo4g。第二个 bean 生成的队列名以 something- 为前缀，后跟 UUID 的 base64 表示。第三个 bean 只使用 UUID 生成名称（不做 base64 转换）——例如 f20c818a-006b-4416-bf91-643590fedb0e。

base64 编码使用 RFC 4648 中的“URL 和文件名安全字母表”。尾部填充字符（=）会被移除。

你可以提供自己的命名策略，从而在队列名中包含其他信息（例如应用名或客户端主机）。

使用 XML 配置时可以指定命名策略。<rabbit:queue> 元素上有 naming-strategy 属性，用于引用实现 AnonymousQueue.NamingStrategy 的 bean。以下示例展示了以各种方式指定命名策略：

<rabbit:queue id="uuidAnon" />

<rabbit:queue id="springAnon" naming-strategy="uuidNamer" />

<rabbit:queue id="customAnon" naming-strategy="customNamer" />

<bean id="uuidNamer" class="org.springframework.amqp.core.AnonymousQueue.UUIDNamingStrategy" />

<bean id="customNamer" class="org.springframework.amqp.core.AnonymousQueue.Base64UrlNamingStrategy">
    <constructor-arg value="custom.gen-" />
</bean>

第一个示例创建如 spring.gen-MRBv9sqISkuCiPfOYfpo4g 这样的名称。第二个示例创建带 UUID 字符串表示的名称。第三个示例创建如 custom.gen-MRBv9sqISkuCiPfOYfpo4g 这样的名称。

你也可以提供自己的命名策略 bean。

从 2.1 版本开始，匿名队列声明时默认把参数 Queue.X_QUEUE_LEADER_LOCATOR 设置为 client-local。这确保队列声明在应用所连接的节点上。你可以在构造实例后调用 queue.setLeaderLocator(null) 恢复以前行为。

恢复自动删除声明

通常，RabbitAdmin 只恢复在应用上下文中声明为 bean 的队列/交换机/绑定；如果任何此类声明是自动删除的，连接丢失时它们会被 broker 移除。连接重新建立时，admin 会重新声明这些实体。通常，通过调用 admin.declareQueue(…​)、admin.declareExchange(…​) 和 admin.declareBinding(…​) 创建的实体不会被恢复。

从 2.4 版本开始，admin 有一个新属性 redeclareManualDeclarations；为 true 时，除应用上下文中的 bean 外，admin 还会恢复这些实体。

如果调用了 deleteQueue(…​)、deleteExchange(…​) 或 removeBinding(…​)，将不会执行单个声明的恢复。删除队列和交换机时，相关绑定会从可恢复实体中移除。

最后，调用 resetAllManualDeclarations() 将阻止恢复任何先前声明的实体。