此版本仍在开发中,尚未被视为稳定版本。对于最新的稳定版本,请使用 Spring Integration 6.4.0spring-doc.cn

AMQP 支持的消息通道

有两种消息通道实现可用。 一个是点对点的,另一个是发布-订阅。 这两个通道都为底层 和 (如本章前面所示的通道适配器和网关) 提供了广泛的配置属性。 但是,我们在此处显示的示例具有最少的配置。 浏览 XML 架构以查看可用属性。AmqpTemplateSimpleMessageListenerContainerspring-doc.cn

点对点通道可能类似于以下示例:spring-doc.cn

<int-amqp:channel id="p2pChannel"/>

在幕后,前面的示例导致声明 named ,并且此通道发送到该 (从技术上讲,通过使用与 this 的名称匹配的路由密钥发送到 no-name 直接交换)。 此通道还会在该 上注册一个 consumer。 如果您希望通道是 “pollable” 而不是消息驱动的,请为标志提供值 ,如下例所示:Queuesi.p2pChannelQueueQueueQueuemessage-drivenfalsespring-doc.cn

<int-amqp:channel id="p2pPollableChannel"  message-driven="false"/>

发布-订阅通道可能如下所示:spring-doc.cn

<int-amqp:publish-subscribe-channel id="pubSubChannel"/>

在幕后,前面的示例会导致声明一个名为 fanout exchange,并且此通道发送到该 fanout 交换。 此通道还声明一个名为 exclusive、auto-delete、non-durable 的服务器,并将其绑定到扇出交换,同时在该交换上注册使用者以接收消息。 publish-subscribe-channel 没有 “pollable” 选项。 它必须是消息驱动的。si.fanout.pubSubChannelQueueQueuespring-doc.cn

从版本 4.1 开始,AMQP 支持的消息通道(与 一起)支持将 和 对于 . 请注意,以前是默认的。 现在,默认情况下,它适用于 .channel-transactedtemplate-channel-transactedtransactionalAbstractMessageListenerContainerRabbitTemplatechannel-transactedtruefalseAbstractMessageListenerContainerspring-doc.cn

在版本 4.3 之前,AMQP 支持的通道仅支持带有有效负载和 Headers 的消息。 整个消息被转换(序列化)并发送到 RabbitMQ。 现在,您可以将属性(或使用 Java 配置时)设置为 . 当此标志为 时,将转换消息有效负荷并映射报头,其方式类似于使用通道适配器时。 这种安排允许 AMQP 支持的通道与不可序列化的有效负载一起使用(可能与另一个消息转换器一起使用,例如 )。 有关默认映射 Headers 的更多信息,请参阅 AMQP Message Headers。 您可以通过提供使用 和 属性的自定义映射器来修改映射。 现在,您还可以指定 ,该 用于在没有标头时设置投放模式。 默认情况下, Spring AMQP 使用交付模式。Serializableextract-payloadsetExtractPayload()truetrueJackson2JsonMessageConverteroutbound-header-mapperinbound-header-mapperdefault-delivery-modeamqp_deliveryModeMessagePropertiesPERSISTENTspring-doc.cn

与其他支持持久性的通道一样,AMQP 支持的通道旨在提供消息持久性以避免消息丢失。 它们不用于将工作分发给其他对等应用程序。 为此,请改用通道适配器。
从版本 5.0 开始,pollable 通道现在阻塞指定 (默认为 1 秒) 的 poller 线程。 以前,与其他实现不同,如果没有可用的消息,则线程会立即返回到调度程序,而不管接收超时如何。 阻塞比使用 a 检索消息(无超时)要贵一些,因为必须创建一个使用者来接收每条消息。 要恢复以前的行为,请将 poller's 设置为 0。receiveTimeoutPollableChannelbasicGet()receiveTimeout

使用 Java 配置进行配置

以下示例显示如何使用 Java 配置配置通道:spring-doc.cn

@Bean
public AmqpChannelFactoryBean pollable(ConnectionFactory connectionFactory) {
    AmqpChannelFactoryBean factoryBean = new AmqpChannelFactoryBean();
    factoryBean.setConnectionFactory(connectionFactory);
    factoryBean.setQueueName("foo");
    factoryBean.setPubSub(false);
    return factoryBean;
}

@Bean
public AmqpChannelFactoryBean messageDriven(ConnectionFactory connectionFactory) {
    AmqpChannelFactoryBean factoryBean = new AmqpChannelFactoryBean(true);
    factoryBean.setConnectionFactory(connectionFactory);
    factoryBean.setQueueName("bar");
    factoryBean.setPubSub(false);
    return factoryBean;
}

@Bean
public AmqpChannelFactoryBean pubSub(ConnectionFactory connectionFactory) {
    AmqpChannelFactoryBean factoryBean = new AmqpChannelFactoryBean(true);
    factoryBean.setConnectionFactory(connectionFactory);
    factoryBean.setQueueName("baz");
    factoryBean.setPubSub(false);
    return factoryBean;
}

使用 Java DSL 进行配置

以下示例显示如何使用 Java DSL 配置通道:spring-doc.cn

@Bean
public IntegrationFlow pollableInFlow(ConnectionFactory connectionFactory) {
    return IntegrationFlow.from(...)
            ...
            .channel(Amqp.pollableChannel(connectionFactory)
                    .queueName("foo"))
            ...
            .get();
}

@Bean
public IntegrationFlow messageDrivenInFow(ConnectionFactory connectionFactory) {
    return IntegrationFlow.from(...)
            ...
            .channel(Amqp.channel(connectionFactory)
                    .queueName("bar"))
            ...
            .get();
}

@Bean
public IntegrationFlow pubSubInFlow(ConnectionFactory connectionFactory) {
    return IntegrationFlow.from(...)
            ...
            .channel(Amqp.publishSubscribeChannel(connectionFactory)
                    .queueName("baz"))
            ...
            .get();
}