Spring 集成提供了通道适配器，用于使用高级消息队列协议 （AMQP） 接收和发送消息。

您需要将此依赖项包含在项目中：

(资料图片仅供参考)

    org.springframework.integration    spring-integration-amqp    6.0.0

以下适配器可用：

Spring 集成还提供了点对点消息通道和由 AMQP 交换和队列支持的发布-订阅消息通道。

为了提供AMQP支持，Spring Integration依赖于（Spring AMQP），它将Spring的核心概念应用于基于AMQP的消息传递解决方案的开发。 Spring AMQP提供了与Spring JMS类似的语义。

虽然提供的AMQP通道适配器仅用于单向消息传递（发送或接收），但Spring Integration还提供了用于请求-回复操作的入站和出站AMQP网关。

提示： 您应该熟悉Spring AMQP项目的参考文档。 它提供了有关Spring与AMQP集成的更深入的信息，特别是RabbitMQ。

入站通道适配器

以下清单显示了 AMQP 入站通道适配器的可能配置选项：

@Beanpublic IntegrationFlow amqpInbound(ConnectionFactory connectionFactory) {    return IntegrationFlow.from(Amqp.inboundAdapter(connectionFactory, "aName"))            .handle(m -> System.out.println(m.getPayload()))            .get();}

从版本 5.5 开始，可以使用在内部调用重试操作时使用的策略进行配置。 有关更多信息，请参阅 JavaDocs。​​AmqpInboundChannelAdapter​​​​org.springframework.amqp.rabbit.retry.MessageRecoverer​​​​RecoveryCallback​​​​setMessageRecoverer()​​

批处理消息

有关批处理消息的更多信息，请参阅Spring AMQP 文档。

要使用 Spring 集成生成批处理消息，只需使用 .​​BatchingRabbitTemplate​​

接收批处理消息时，默认情况下，侦听器容器提取每个片段消息，适配器将为每个片段生成 。 从版本 5.2 开始，如果容器的属性设置为 ，则由适配器执行去批处理，并生成一个有效负载为片段有效负载列表（如果适用，转换后）。​​Message​​​​deBatchingEnabled​​​​false​​​​Message>​​

默认值为 ，但可以在适配器上覆盖。​​BatchingStrategy​​​​SimpleBatchingStrategy​​

轮询入站通道适配器

概述

版本 5.0.1 引入了轮询通道适配器，允许您按需获取单个消息 — 例如，使用 或 轮询器。 有关详细信息，请参阅延迟确认可轮询消息源。​​MessageSourcePollingTemplate​​

它当前不支持 XML 配置。

以下示例显示如何配置：​​AmqpMessageSource​​

@Beanpublic IntegrationFlow flow() {    return IntegrationFlow.from(Amqp.inboundPolledAdapter(connectionFactory(), DSL_QUEUE),                    e -> e.poller(Pollers.fixedDelay(1_000)).autoStartup(false))            .handle(p -> {                ...            })            .get();}

有关配置属性，请参阅Javadoc。

批处理消息

请参阅批处理消息。

对于轮询适配器，没有侦听器容器，批处理消息始终是反批处理的（如果支持这样做）。​​BatchingStrategy​​

入站网关

入站网关支持入站通道适配器上的所有属性（除了“通道”被“请求通道”替换），以及一些其他属性。 以下清单显示了可用的属性：

@Bean // return the upper cased payloadpublic IntegrationFlow amqpInboundGateway(ConnectionFactory connectionFactory) {    return IntegrationFlow.from(Amqp.inboundGateway(connectionFactory, "foo"))            .transform(String.class, String::toUpperCase)            .get();}

请参阅入站通道适配器中有关配置属性的说明。​​listener-container​​

从版本 5.5 开始，可以使用在内部调用重试操作时使用的策略进行配置。 有关更多信息，请参阅 JavaDocs。​​AmqpInboundChannelAdapter​​​​org.springframework.amqp.rabbit.retry.MessageRecoverer​​​​RecoveryCallback​​​​setMessageRecoverer()​​

批处理消息

请参阅批处理消息。

入站终端节点确认模式

默认情况下，入站终端节点使用应答方式，这意味着容器会在下游集成流完成（或使用 或 将消息传递给另一个线程）时自动确认消息。 将模式设置为配置使用者，以便根本不使用确认（代理在发送消息后立即自动确认消息）。 设置模式以允许用户代码在处理过程中的某个其他点确认消息。 为了支持此功能，在此模式下，终结点分别在 和 中提供 和 标头。​​AUTO​​​​QueueChannel​​​​ExecutorChannel​​​​NONE​​​​MANUAL​​​​Channel​​​​deliveryTag​​​​amqp_channel​​​​amqp_deliveryTag​​

您可以对 执行任何有效的 Rabbit 命令，但通常只使用 and（或）。 为了不干扰容器的操作，不应保留对通道的引用，而应仅在当前消息的上下文中使用它。​​Channel​​​​basicAck​​​​basicNack​​​​basicReject​​

以下示例演示如何使用确认：​​MANUAL​​

@ServiceActivator(inputChannel = "foo", outputChannel = "bar")public Object handle(@Payload String payload, @Header(AmqpHeaders.CHANNEL) Channel channel,        @Header(AmqpHeaders.DELIVERY_TAG) Long deliveryTag) throws Exception {    // Do some processing    if (allOK) {        channel.basicAck(deliveryTag, false);        // perhaps do some more processing    }    else {        channel.basicNack(deliveryTag, false, true);    }    return someResultForDownStreamProcessing;}

出站终结点

以下出站终结点具有许多类似的配置选项。 从版本 5.2 开始，已添加。 通常，当启用发布者确认时，代理将快速返回一个 ack（或 nack），该 ack（或 nack）将被发送到相应的通道。 如果在收到确认之前关闭了通道，Spring AMQP 框架将合成一个 nack。 “丢失”确认不应发生，但如果设置此属性，则终结点将定期检查它们，并在时间过去而未收到确认时合成 nack。​​confirm-timeout​​

出站通道适配器

以下示例显示了 AMQP 出站通道适配器的可用属性：

@Beanpublic IntegrationFlow amqpOutbound(AmqpTemplate amqpTemplate,        MessageChannel amqpOutboundChannel) {    return IntegrationFlow.from(amqpOutboundChannel)            .handle(Amqp.outboundAdapter(amqpTemplate)                        .routingKey("queue1")) // default exchange - route to queue "queue1"            .get();}

出站网关

以下清单显示了 AMQP 出站网关的可能属性：

@Beanpublic IntegrationFlow amqpOutbound(AmqpTemplate amqpTemplate) {    return f -> f.handle(Amqp.outboundGateway(amqpTemplate)                    .routingKey("foo")) // default exchange - route to queue "foo"            .get();}@MessagingGateway(defaultRequestChannel = "amqpOutbound.input")public interface MyGateway {    String sendToRabbit(String data);}

请注意，出站适配器和出站网关配置之间的唯一区别是属性的设置。​​expectReply​​

异步出站网关

上一节中讨论的网关是同步的，因为发送线程挂起，直到 收到回复（或发生超时）。 Spring Integration 版本 4.3 添加了一个异步网关，该网关使用 from Spring AMQP。 发送消息时，线程会在发送操作完成后立即返回，收到消息后，将在模板的侦听器容器线程上发送回复。 在轮询器线程上调用网关时，这可能很有用。 线程已释放，可用于框架中的其他任务。​​AsyncRabbitTemplate​​

以下清单显示了 AMQP 异步出站网关的可能配置选项：

@Configurationpublic class AmqpAsyncApplication {    @Bean    public IntegrationFlow asyncAmqpOutbound(AsyncRabbitTemplate asyncRabbitTemplate) {        return f -> f                .handle(Amqp.asyncOutboundGateway(asyncRabbitTemplate)                        .routingKey("queue1")); // default exchange - route to queue "queue1"    }    @MessagingGateway(defaultRequestChannel = "asyncAmqpOutbound.input")    public interface MyGateway {        String sendToRabbit(String data);    }}

另请参阅异步服务激活器以获取详细信息。

发布商确认和返回的替代机制

将连接工厂配置为发布者确认并返回时，上述部分将讨论消息通道的配置，以便异步接收确认和返回。 从版本 5.4 开始，还有一个通常更易于使用的附加机制。

在这种情况下，请勿配置 或 确认和返回通道。 相反，在标头中添加一个实例;然后，您可以通过检查已发送消息的实例中的未来状态来等待稍后的结果。 在将来完成之前，将始终填充该字段（如果返回消息）。​​confirm-correlation-expression​​​​CorrelationData​​​​AmqpHeaders.PUBLISH_CONFIRM_CORRELATION​​​​CorrelationData​​​​returnedMessage​​

CorrelationData corr = new CorrelationData("someId"); // <--- Unique "id" is required for returnssomeFlow.getInputChannel().send(MessageBuilder.withPayload("test")        .setHeader("rk", "someKeyThatWontRoute")        .setHeader(AmqpHeaders.PUBLISH_CONFIRM_CORRELATION, corr)        .build());...try {    Confirm Confirm = corr.getFuture().get(10, TimeUnit.SECONDS);    Message returned = corr.getReturnedMessage();    if (returned !- null) {        // message could not be routed    }}catch { ... }

为了提高性能，您可能希望发送多条消息并稍后等待确认，而不是一次发送一条。 返回的消息是转换后的原始消息;您可以使用所需的任何其他数据对 A 进行子类。​​CorrelationData​​

入站消息转换

到达通道适配器或网关的入站消息将使用消息转换器转换为有效负载。 默认情况下，使用 a，用于处理 java 序列化和文本。 默认情况下，标头使用 映射。 如果发生转换错误，并且未定义错误通道，则会将异常引发到容器，并由侦听器容器的错误处理程序处理。 默认错误处理程序将转换错误视为致命错误，消息将被拒绝（如果队列已如此配置，则路由到死信交换）。 如果定义了错误通道，则有效负载为 具有属性（无法转换的 Spring AMQP 消息）和 . 如果容器是（默认值），并且错误流使用错误而不引发异常，则将确认原始消息。 如果错误流引发异常，则异常类型与容器的错误处理程序一起确定消息是否重新排队。 如果容器配置了 ，则有效负载是具有附加属性和 的 。 这使错误流能够调用 或（或）消息，以控制其处置。​​spring-messaging​​​​Message​​​​SimpleMessageConverter​​​​DefaultHeaderMapper.inboundMapper()​​​​ErrorMessage​​​​ListenerExecutionFailedException​​​​failedMessage​​​​cause​​​​AcknowledgeMode​​​​AUTO​​​​AcknowledgeMode.MANUAL​​​​ManualAckListenerExecutionFailedException​​​​channel​​​​deliveryTag​​​​basicAck​​​​basicNack​​​​basicReject​​

出站消息转换

Spring AMQP 1.4 引入了 ，其中实际转换器的选择基于 在传入内容类型消息属性上。 这可由入站终端节点使用。​​ContentTypeDelegatingMessageConverter​​

从 Spring 集成版本 4.3 开始，您也可以在出站端点上使用 ，标头指定使用哪个转换器。​​ContentTypeDelegatingMessageConverter​​​​contentType​​

以下示例配置了一个 ，默认转换器为 （处理 Java 序列化和纯文本），以及一个 JSON 转换器：​​ContentTypeDelegatingMessageConverter​​​​SimpleMessageConverter​​

将标头设置为 to 的消息发送到会导致选择 JSON 转换器。​​ctRequestChannel​​​​contentType​​​​application/json​​

这适用于出站通道适配器和网关。

出站用户标识

Spring AMQP 版本 1.6 引入了一种机制，允许为出站消息指定默认用户 ID。 始终可以设置标头，该标头现在优先于默认值。 这可能对邮件收件人有用。 对于入站邮件，如果邮件发布者设置了该属性，则该属性将在标头中可用。 请注意，RabbitMQ会验证用户 ID 是连接的实际用户 ID，还是连接允许模拟。​​AmqpHeaders.USER_ID​​​​AmqpHeaders.RECEIVED_USER_ID​​

要为出站消息配置缺省用户标识，请在 上配置该标识，并将出站适配器或网关配置为使用该模板。 同样，要在回复上设置用户 ID 属性，请将适当配置的模板注入入站网关。 有关更多信息，请参阅Spring AMQP 文档。​​RabbitTemplate​​

延迟消息交换

Spring AMQP 支持RabbitMQ 延迟消息交换插件。 对于入站邮件，标头映射到标头。 设置标头会导致在出站邮件中设置相应的标头。 还可以在出站终结点上指定 and 属性（使用 XML 配置时）。 这些属性优先于标头。​​x-delay​​​​AmqpHeaders.RECEIVED_DELAY​​​​AMQPHeaders.DELAY​​​​x-delay​​​​delay​​​​delayExpression​​​​delay-expression​​​​AmqpHeaders.DELAY​​

AMQP 支持的消息通道

有两种消息通道实现可用。 一个是点对点，另一个是发布-订阅。 这两个通道都为基础和（如本章前面的通道适配器和网关所示）提供了广泛的配置属性。 但是，我们在此处显示的示例具有最小配置。 浏览 XML 架构以查看可用属性。​​AmqpTemplate​​​​SimpleMessageListenerContainer​​

点对点通道可能类似于以下示例：

在幕后，前面的示例导致声明一个命名，并且此通道发送到该通道（从技术上讲，通过使用与此名称匹配的路由密钥发送到无名称的直接交换）。 此通道还在此 上注册消费者。 如果希望通道是“可轮询的”而不是消息驱动的，请为标志提供值 ，如以下示例所示：​​Queue​​​​si.p2pChannel​​​​Queue​​​​Queue​​​​Queue​​​​message-driven​​​​false​​

发布-订阅通道可能如下所示：

在后台，前面的示例会导致声明名为的扇出交换，并且此通道发送到该扇出交换。 此通道还声明一个以服务器命名的独占、自动删除、非持久，并将其绑定到扇出交换，同时注册使用者以接收消息。 发布-订阅-通道没有“可轮询”选项。 它必须是消息驱动的。​​si.fanout.pubSubChannel​​​​Queue​​​​Queue​​

从版本 4.1 开始，AMQP 支持的消息通道（与 一起）支持单独配置 和 对于 . 请注意，以前是默认的。 现在，默认情况下，它适用于 .​​channel-transacted​​​​template-channel-transacted​​​​transactional​​​​AbstractMessageListenerContainer​​​​RabbitTemplate​​​​channel-transacted​​​​true​​​​false​​​​AbstractMessageListenerContainer​​

在版本 4.3 之前，AMQP 支持的通道仅支持具有有效负载和标头的消息。 整个消息被转换（序列化）并发送到 RabbitMQ。 现在，您可以将属性（或使用 Java 配置时）设置为 。 当此标志为 时，将转换消息有效负载并映射标头，其方式类似于使用通道适配器时的方式。 这种安排允许 AMQP 支持的通道与不可序列化的有效负载一起使用（可能与其他消息转换器一起使用，例如 ）。 有关默认映射标头的更多信息，请参阅AMQP 消息标头。 您可以通过提供使用 和 属性的自定义映射器来修改映射。 现在，您还可以指定 ，用于在没有标头时设置传递模式。 默认情况下，Spring AMQP 使用交付模式。​​Serializable​​​​extract-payload​​​​setExtractPayload()​​​​true​​​​true​​​​Jackson2JsonMessageConverter​​​​outbound-header-mapper​​​​inbound-header-mapper​​​​default-delivery-mode​​​​amqp_deliveryMode​​​​MessageProperties​​​​PERSISTENT​​

使用 Java 配置进行配置

以下示例显示如何使用 Java 配置配置通道：

@Beanpublic AmqpChannelFactoryBean pollable(ConnectionFactory connectionFactory) {    AmqpChannelFactoryBean factoryBean = new AmqpChannelFactoryBean();    factoryBean.setConnectionFactory(connectionFactory);    factoryBean.setQueueName("foo");    factoryBean.setPubSub(false);    return factoryBean;}@Beanpublic AmqpChannelFactoryBean messageDriven(ConnectionFactory connectionFactory) {    AmqpChannelFactoryBean factoryBean = new AmqpChannelFactoryBean(true);    factoryBean.setConnectionFactory(connectionFactory);    factoryBean.setQueueName("bar");    factoryBean.setPubSub(false);    return factoryBean;}@Beanpublic AmqpChannelFactoryBean pubSub(ConnectionFactory connectionFactory) {    AmqpChannelFactoryBean factoryBean = new AmqpChannelFactoryBean(true);    factoryBean.setConnectionFactory(connectionFactory);    factoryBean.setQueueName("baz");    factoryBean.setPubSub(false);    return factoryBean;}

使用 Java DSL 进行配置

以下示例显示如何使用 Java DSL 配置通道：

@Beanpublic IntegrationFlow pollableInFlow(ConnectionFactory connectionFactory) {    return IntegrationFlow.from(...)            ...            .channel(Amqp.pollableChannel(connectionFactory)                    .queueName("foo"))            ...            .get();}@Beanpublic IntegrationFlow messageDrivenInFow(ConnectionFactory connectionFactory) {    return IntegrationFlow.from(...)            ...            .channel(Amqp.channel(connectionFactory)                    .queueName("bar"))            ...            .get();}@Beanpublic IntegrationFlow pubSubInFlow(ConnectionFactory connectionFactory) {    return IntegrationFlow.from(...)            ...            .channel(Amqp.publishSubscribeChannel(connectionFactory)                    .queueName("baz"))            ...            .get();}

AMQP 消息标头

概述

Spring 集成 AMQP 适配器会自动映射所有 AMQP 属性和标头。 （这是对 4.3 的更改 - 以前，仅映射标准标头）。 默认情况下，这些属性通过使用DefaultAmqpHeaderMapper复制到 Spring Integration 中。MessageHeaders

您可以传入自己的特定于 AMQP 的标头映射程序的实现，因为适配器具有支持这样做的属性。

AMQP 消息属性中的任何用户定义的标头都将复制到 AMQP 消息或从 AMQP 消息复制，除非 的 或属性显式否定。 默认情况下，对于出站映射器，不映射任何标头。 请参阅本节后面出现的警告，了解原因。​​requestHeaderNames​​​​replyHeaderNames​​​​DefaultAmqpHeaderMapper​​​​x-*​​

要覆盖默认值并恢复到 4.3 之前的行为，请在属性中使用 和。​​STANDARD_REQUEST_HEADERS​​​​STANDARD_REPLY_HEADERS​​

从版本 4.1 开始，（超类）允许为 和 属性（除了现有的 和 ）配置令牌，以映射所有用户定义的标头。​​AbstractHeaderMapper​​​​DefaultAmqpHeaderMapper​​​​NON_STANDARD_HEADERS​​​​requestHeaderNames​​​​replyHeaderNames​​​​STANDARD_REQUEST_HEADERS​​​​STANDARD_REPLY_HEADERS​​

该类标识由 ：​​org.springframework.amqp.support.AmqpHeaders​​​​DefaultAmqpHeaderMapper​​

从版本 4.3 开始，标头映射中的模式可以通过在模式前面加上 来否定。 否定模式获得优先级，因此诸如 不映射（也不是 ）之类的列表。 标准标头加上 和 被映射。 否定技术可能很有用，例如，当 JSON 反序列化逻辑以不同的方式在接收方下游完成时，不映射传入消息的 JSON 类型标头。 为此，应为入站通道适配器/网关的标头映射器配置模式。​​!​​​​STANDARD_REQUEST_HEADERS,thing1,ba*,!thing2,!thing3,qux,!thing1​​​​thing1​​​​thing2​​​​thing3​​​​bad​​​​qux​​​​!json_*​​

页眉​​contentType​​

与其他标头不同，不以 ;这允许跨不同技术透明地传递 contentType 标头。 例如，发送到 RabbitMQ 队列的入站 HTTP 消息。​​AmqpHeaders.CONTENT_TYPE​​​​amqp_​​

标头映射到Spring AMQP的属性，随后映射到RabbitMQ的属性。​​contentType​​​​MessageProperties.contentType​​​​content_type​​

在版本 5.1 之前，此标头也映射为映射中的条目;这是不正确的，此外，该值可能是错误的，因为基础 Spring AMQP 消息转换器可能已更改内容类型。 这样的更改将反映在 first-class 属性中，但不反映在 RabbitMQ 标头映射中。 入站映射忽略标头映射值。 不再映射到标头映射中的条目。​​MessageProperties.headers​​​​content_type​​​​contentType​​

严格的消息排序

本节介绍入站和出站消息的消息排序。

入境

如果需要对入站消息进行严格排序，则必须将入站侦听器容器的属性配置为 。 这是因为，如果消息失败并重新传递，它将在现有的预取消息之后到达。 从 Spring AMQP 2.0 版本开始，默认为提高性能。 严格的订购要求是以性能下降为代价的。​​prefetchCount​​​​1​​​​prefetchCount​​​​250​​

出境

请考虑以下集成流程：

@Beanpublic IntegrationFlow flow(RabbitTemplate template) {    return IntegrationFlow.from(Gateway.class)            .split(s -> s.delimiters(","))            .transform(String::toUpperCase)            .handle(Amqp.outboundAdapter(template).routingKey("rk"))            .get();}

假设我们发送消息 ，并发送到网关。 虽然消息 、 很可能是按顺序发送的，但不能保证。 这是因为模板为每个发送操作从缓存中“借用”一个通道，并且不能保证对每条消息使用相同的通道。 一种解决方案是在拆分器之前启动事务，但是在 RabbitMQ 中事务成本高昂，并且会使性能降低数百倍。​​A​​​​B​​​​C​​​​A​​​​B​​​​C​​

为了以更有效的方式解决这个问题，从版本 5.1 开始，Spring 集成提供了 这是一个 . 请参阅处理消息建议。 在拆分器之前应用时，它可确保在同一通道上执行所有下游操作，并且可以选择等到收到所有已发送消息的发布者确认（如果连接工厂配置为确认）。 以下示例演示如何使用：​​BoundRabbitChannelAdvice​​​​HandleMessageAdvice​​​​BoundRabbitChannelAdvice​​

@Beanpublic IntegrationFlow flow(RabbitTemplate template) {    return IntegrationFlow.from(Gateway.class)            .split(s -> s.delimiters(",")                    .advice(new BoundRabbitChannelAdvice(template, Duration.ofSeconds(10))))            .transform(String::toUpperCase)            .handle(Amqp.outboundAdapter(template).routingKey("rk"))            .get();}

请注意，在建议和出站适配器中使用了相同的（实现）。 该建议在模板的方法中运行下游流，以便所有操作都在同一通道上运行。 如果提供了可选的超时，则当流完成时，建议将调用该方法，如果在指定时间内未收到确认，则会引发异常。​​RabbitTemplate​​​​RabbitOperations​​​​invoke​​​​waitForConfirmsOrDie​​

AMQP 样品

要试验 AMQP 适配器，请查看 Spring 集成示例 git 存储库中提供的示例，网址为https://github.com/SpringSource/spring-integration-samples

目前，一个示例通过使用出站通道适配器和入站通道适配器演示了 Spring 集成 AMQP 适配器的基本功能。 由于示例中的 AMQP 代理实现使用RabbitMQ。

启动示例应用程序后，在命令提示符下输入一些文本，包含该输入文本的消息将调度到 AMQP 队列。 作为回报，Spring Integration 检索该消息并将其打印到控制台。

下图说明了此示例中使用的一组基本 Spring 集成组件。

RabbitMQ 流队列支持

AMQP 示例图像的弹簧集成图：：images/spring-integration-amqp-sample-graph.png[]

版本 6.0 引入了对 RabbitMQ 流队列的支持。

这些终结点的 DSL 工厂类是 。​​Rabbit​​

RabbitMQ 流入站通道适配器

@BeanIntegrationFlow flow(Environment env) {    @Bean    IntegrationFlow simpleStream(Environment env) {        return IntegrationFlow.from(RabbitStream.inboundAdapter(env)                        .configureContainer(container -> container.queueName("my.stream")))                // ...                .get();    }    @Bean    IntegrationFlow superStream(Environment env) {        return IntegrationFlow.from(RabbitStream.inboundAdapter(env)                        .configureContainer(container -> container.superStream("my.stream", "my.consumer")))                // ...                .get();    }}

RabbitMQ 流出站通道适配器

@BeanIntegrationFlow outbound(RabbitStreamTemplate template) {    return f -> f            // ...            .handle(RabbitStream.outboundStreamAdapter(template));}