(window.webpackJsonp=window.webpackJsonp||[]).push([[456],{887:function(e,t,n){"use strict";n.r(t);var r=n(56),v=Object(r.a)({},(function(){var e=this,t=e.$createElement,n=e._self._c||t;return n("ContentSlotsDistributor",{attrs:{"slot-key":e.$parent.slotKey}},[n("h1",{attrs:{id:"amqp-支持"}},[n("a",{staticClass:"header-anchor",attrs:{href:"#amqp-支持"}},[e._v("#")]),e._v(" AMQP 支持")]),e._v(" "),n("h2",{attrs:{id:"amqp-支持-2"}},[n("a",{staticClass:"header-anchor",attrs:{href:"#amqp-支持-2"}},[e._v("#")]),e._v(" AMQP 支持")]),e._v(" "),n("p",[e._v("Spring 集成提供了用于通过使用高级消息队列协议接收和发送消息的通道适配器。")]),e._v(" "),n("p",[e._v("你需要在项目中包含此依赖项:")]),e._v(" "),n("p",[e._v("Maven")]),e._v(" "),n("div",{staticClass:"language- extra-class"},[n("pre",{pre:!0,attrs:{class:"language-text"}},[n("code",[e._v("\n org.springframework.integration\n spring-integration-amqp\n 5.5.9\n\n")])])]),n("p",[e._v("Gradle")]),e._v(" "),n("div",{staticClass:"language- extra-class"},[n("pre",{pre:!0,attrs:{class:"language-text"}},[n("code",[e._v('compile "org.springframework.integration:spring-integration-amqp:5.5.9"\n')])])]),n("p",[e._v("可提供以下适配器:")]),e._v(" "),n("ul",[n("li",[n("p",[n("a",{attrs:{href:"#amqp-inbound-channel-adapter"}},[e._v("入站通道适配器")])])]),e._v(" "),n("li",[n("p",[n("a",{attrs:{href:"#amqp-inbound-gateway"}},[e._v("入站网关")])])]),e._v(" "),n("li",[n("p",[n("a",{attrs:{href:"#amqp-outbound-channel-adapter"}},[e._v("出站通道适配器")])])]),e._v(" "),n("li",[n("p",[n("a",{attrs:{href:"#amqp-outbound-gateway"}},[e._v("出站网关")])])]),e._v(" "),n("li",[n("p",[n("a",{attrs:{href:"#amqp-async-outbound-gateway"}},[e._v("异步出站网关")])])])]),e._v(" "),n("p",[e._v("Spring 集成还提供了由 AMQP 交换和队列支持的点对点消息通道和发布-订阅消息通道。")]),e._v(" "),n("p",[e._v("为了提供 AMQP 支持, Spring 集成依赖于("),n("a",{attrs:{href:"https://projects.spring.io/spring-amqp",target:"_blank",rel:"noopener noreferrer"}},[e._v("Spring AMQP"),n("OutboundLink")],1),e._v("),它将核心 Spring 概念应用于基于 AMQP 的消息传递解决方案的开发。 Spring AMQP 提供了与("),n("a",{attrs:{href:"https://docs.spring.io/spring/docs/current/spring-framework-reference/integration.html#jms",target:"_blank",rel:"noopener noreferrer"}},[e._v("Spring JMS"),n("OutboundLink")],1),e._v(")类似的语义。")]),e._v(" "),n("p",[e._v("Spring 鉴于所提供的 AMQP 通道适配器仅用于单向消息传递(发送或接收), Spring 集成还提供了用于请求-应答操作的入站和出站 AMQP 网关。")]),e._v(" "),n("p",[e._v("小贴士:你应该熟悉"),n("a",{attrs:{href:"https://docs.spring.io/spring-amqp/reference/html/",target:"_blank",rel:"noopener noreferrer"}},[e._v("reference documentation of the Spring AMQP project"),n("OutboundLink")],1),e._v("。它提供了关于 Spring 与 AMQP 的集成的更深入的信息,特别是与 RabbitMQ 的集成。")]),e._v(" "),n("h3",{attrs:{id:"入站通道适配器"}},[n("a",{staticClass:"header-anchor",attrs:{href:"#入站通道适配器"}},[e._v("#")]),e._v(" 入站通道适配器")]),e._v(" "),n("p",[e._v("下面的清单显示了 AMQP 入站通道适配器的可能配置选项:")]),e._v(" "),n("p",[e._v("爪哇 DSL")]),e._v(" "),n("div",{staticClass:"language- extra-class"},[n("pre",{pre:!0,attrs:{class:"language-text"}},[n("code",[e._v('@Bean\npublic IntegrationFlow amqpInbound(ConnectionFactory connectionFactory) {\n return IntegrationFlows.from(Amqp.inboundAdapter(connectionFactory, "aName"))\n .handle(m -> System.out.println(m.getPayload()))\n .get();\n}\n')])])]),n("p",[e._v("爪哇")]),e._v(" "),n("div",{staticClass:"language- extra-class"},[n("pre",{pre:!0,attrs:{class:"language-text"}},[n("code",[e._v('@Bean\npublic MessageChannel amqpInputChannel() {\n return new DirectChannel();\n}\n\n@Bean\npublic AmqpInboundChannelAdapter inbound(SimpleMessageListenerContainer listenerContainer,\n @Qualifier("amqpInputChannel") MessageChannel channel) {\n AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(listenerContainer);\n adapter.setOutputChannel(channel);\n return adapter;\n}\n\n@Bean\npublic SimpleMessageListenerContainer container(ConnectionFactory connectionFactory) {\n SimpleMessageListenerContainer container =\n new SimpleMessageListenerContainer(connectionFactory);\n container.setQueueNames("aName");\n container.setConcurrentConsumers(2);\n // ...\n return container;\n}\n\n@Bean\n@ServiceActivator(inputChannel = "amqpInputChannel")\npublic MessageHandler handler() {\n return new MessageHandler() {\n\n @Override\n public void handleMessage(Message> message) throws MessagingException {\n System.out.println(message.getPayload());\n }\n\n };\n}\n')])])]),n("p",[e._v("XML")]),e._v(" "),n("div",{staticClass:"language- extra-class"},[n("pre",{pre:!0,attrs:{class:"language-text"}},[n("code",[e._v(' (27)\n')])])]),n("table",[n("thead",[n("tr",[n("th",[n("strong",[e._v("1")])]),e._v(" "),n("th",[e._v("此适配器的唯一 ID。"),n("br"),e._v("可选的。")])])]),e._v(" "),n("tbody",[n("tr",[n("td",[n("strong",[e._v("2")])]),e._v(" "),n("td",[e._v("转换后的消息应发送到的消息通道。"),n("br"),e._v("需要。")])]),e._v(" "),n("tr",[n("td",[n("strong",[e._v("3")])]),e._v(" "),n("td",[e._v("AMQP 队列的名称(用逗号分隔的列表),应该从这些队列中消费消息。"),n("br"),e._v("required。")])]),e._v(" "),n("tr",[n("td",[n("strong",[e._v("4")])]),e._v(" "),n("td",[n("code",[e._v("MessageListenerContainer")]),e._v("的确认模式,"),n("br"),e._v("当设置为"),n("code",[e._v("MANUAL")]),e._v("时,传递标记和通道在消息头"),n("code",[e._v("amqp_deliveryTag")]),e._v("和"),n("code",[e._v("amqp_channel")]),e._v("中提供,分别。"),n("br"),e._v("用户应用程序负责确认。"),n("code",[e._v("NONE")]),e._v("表示没有确认("),n("code",[e._v("autoAck")]),e._v(")。"),n("code",[e._v("AUTO")]),e._v("表示适配器的容器在下游流完成时进行确认。"),n("br"),e._v("可选(默认为自动)。"),n("br"),e._v("参见"),n("a",{attrs:{href:"#amqp-inbound-ack"}},[e._v("入站端点确认模式")]),e._v("。")])]),e._v(" "),n("tr",[n("td",[n("strong",[e._v("5")])]),e._v(" "),n("td",[e._v("额外的 AOP 建议来处理与此入站通道适配器相关的横切行为。"),n("br"),e._v("可选的。")])]),e._v(" "),n("tr",[n("td",[n("strong",[e._v("6")])]),e._v(" "),n("td",[e._v("标志,指示该组件创建的通道是事务性的。"),n("br"),e._v("如果为真,它会告诉框架使用事务性通道,并根据结果以提交或回滚结束所有操作(发送或接收),但有一个异常,表示回滚。"),n("br"),e._v("可选(默认为假)。")])]),e._v(" "),n("tr",[n("td",[n("strong",[e._v("7")])]),e._v(" "),n("td",[e._v("指定要创建的并发消费者的数量。"),n("br"),e._v("默认值是"),n("code",[e._v("1")]),e._v("。"),n("br"),e._v("我们建议增加并发消费者的数量,以按比例分配来自队列的消息的消耗,"),n("br"),e._v("但是,请注意,一旦注册了多个消费者,任何订购保证都会丢失。,通常情况下,"),n("br"),e._v(",对于低容量队列使用一个使用者。"),n("br"),e._v("设置了“每个队列的使用者”时不允许使用。"),n("br"),e._v("可选。")])]),e._v(" "),n("tr",[n("td",[n("strong",[e._v("8")])]),e._v(" "),n("td",[e._v("Bean 参考 RabbitMQ.可选(默认为)。")])]),e._v(" "),n("tr",[n("td",[n("strong",[e._v("9")])]),e._v(" "),n("td",[e._v("错误消息应发送到的消息通道。"),n("br"),e._v("可选。")])]),e._v(" "),n("tr",[n("td",[n("strong",[e._v("10")])]),e._v(" "),n("td",[e._v("侦听器通道是否暴露于已注册的"),n("code",[e._v("ChannelAwareMessageListener")]),e._v("."),n("br"),e._v("可选(默认为 true)。")])]),e._v(" "),n("tr",[n("td",[n("strong",[e._v("11")])]),e._v(" "),n("td",[e._v("在接收 AMQP 消息时使用的对"),n("code",[e._v("AmqpHeaderMapper")]),e._v("的引用。"),n("br"),e._v("可选的。"),n("br"),e._v("默认情况下,只有标准的 AMQP 属性(例如"),n("code",[e._v("contentType")]),e._v(")被复制到 Spring Integration"),n("code",[e._v("MessageHeaders")]),e._v("。"),n("br"),e._v("AMQP"),n("code",[e._v("MessageProperties")]),e._v("中的任何用户定义的标题都不会通过默认的"),n("code",[e._v("DefaultAmqpHeaderMapper")]),e._v("复制到消息中。如果提供了“request-header-names”,则不允许"),n("br"),e._v("。")])]),e._v(" "),n("tr",[n("td",[n("strong",[e._v("12")])]),e._v(" "),n("td",[e._v("要从 AMQP 请求映射到"),n("code",[e._v("MessageHeaders")]),e._v("中的 AMQP 头的名称的逗号分隔列表。"),n("br"),e._v("只能提供如果没有提供“header-mapper”引用。"),n("br"),e._v("此列表中的值也可以是与 header 名称匹配的简单模式(例如“*”或“thing1*,thing2”或“*something”)。")])]),e._v(" "),n("tr",[n("td",[n("strong",[e._v("13")])]),e._v(" "),n("td",[e._v("引用"),n("code",[e._v("AbstractMessageListenerContainer")]),e._v("用于接收 AMQP 消息。"),n("br"),e._v("如果提供了此属性,则不应提供与侦听器容器配置相关的其他属性。"),n("br"),e._v("换句话说,通过设置此引用,你必须对侦听器容器配置承担全部责任。"),n("br"),e._v("唯一的例外是"),n("code",[e._v("MessageListener")]),e._v("本身。"),n("br"),e._v("因为这实际上是这个通道适配器实现的核心责任,所以引用的侦听器容器不能已经拥有自己的"),n("code",[e._v("MessageListener")]),e._v("。"),n("br"),e._v("可选的。")])]),e._v(" "),n("tr",[n("td",[n("strong",[e._v("14")])]),e._v(" "),n("td",[e._v("在接收 AMQP 消息时使用的"),n("code",[e._v("MessageConverter")]),e._v("。"),n("br"),e._v("可选。")])]),e._v(" "),n("tr",[n("td",[n("strong",[e._v("15")])]),e._v(" "),n("td",[e._v("在接收 AMQP 消息时使用的"),n("code",[e._v("MessagePropertiesConverter")]),e._v("。"),n("br"),e._v("可选。")])]),e._v(" "),n("tr",[n("td",[n("strong",[e._v("16")])]),e._v(" "),n("td",[e._v("指定底层"),n("code",[e._v("AbstractMessageListenerContainer")]),e._v("应该启动和停止的阶段,"),n("br"),e._v("启动顺序从最低到最高,关闭顺序与此相反,"),n("br"),e._v("默认情况下,该值为"),n("code",[e._v("Integer.MAX_VALUE")]),e._v(",这意味着这个容器开始得越晚,停止得越快。"),n("br"),e._v("可选的。")])]),e._v(" "),n("tr",[n("td",[n("strong",[e._v("17")])]),e._v(" "),n("td",[e._v("告诉 AMQP 代理在一个请求中要向每个使用者发送多少消息。"),n("br"),e._v("通常,你可以将这个值设置得很高,以提高吞吐量。"),n("br"),e._v("它应该大于或等于事务大小(请参见"),n("code",[e._v("tx-size")]),e._v("属性,"),n("br"),e._v("可选(默认为"),n("code",[e._v("1")]),e._v(")。")])]),e._v(" "),n("tr",[n("td",[n("strong",[e._v("18")])]),e._v(" "),n("td",[e._v("接收以毫秒为单位的超时。"),n("br"),e._v("可选(默认为"),n("code",[e._v("1000")]),e._v(")。")])]),e._v(" "),n("tr",[n("td",[n("strong",[e._v("19")])]),e._v(" "),n("td",[e._v("指定底层"),n("code",[e._v("AbstractMessageListenerContainer")]),e._v("的恢复尝试之间的间隔(以毫秒为单位)。"),n("br"),e._v("可选(默认为"),n("code",[e._v("5000")]),e._v(")。")])]),e._v(" "),n("tr",[n("td",[n("strong",[e._v("20")])]),e._v(" "),n("td",[e._v("如果“true”且代理上没有队列可用,则容器在启动时抛出一个致命的异常,如果在容器运行时删除了队列(在进行了三次被动声明队列的尝试之后),则停止。,容器不抛出异常并进入恢复模式,尝试根据"),n("code",[e._v("recovery-interval")]),e._v("重新启动。"),n("br"),e._v("可选(默认为"),n("code",[e._v("true")]),e._v(")。")])]),e._v(" "),n("tr",[n("td",[n("strong",[e._v("21")])]),e._v(" "),n("td",[e._v("在底层"),n("code",[e._v("AbstractMessageListenerContainer")]),e._v("停止之后和强制关闭 AMQP 连接之前等待工作人员的时间(以毫秒为单位)。"),n("br"),e._v("如果有工作人员在关闭信号出现时处于活动状态,只要他们能够在此超时范围内完成处理,他们就被允许完成处理。"),n("br"),e._v("否则,连接已关闭,消息仍未确认(如果通道是事务性的)。"),n("br"),e._v("可选(默认为"),n("code",[e._v("5000")]),e._v(")。")])]),e._v(" "),n("tr",[n("td",[n("strong",[e._v("22")])]),e._v(" "),n("td",[e._v("默认情况下,底层"),n("code",[e._v("AbstractMessageListenerContainer")]),e._v("使用"),n("code",[e._v("SimpleAsyncTaskExecutor")]),e._v("实现,该实现为每个任务启动一个新线程,并异步运行它,默认情况下,>")]),e._v(",其中每个消息都有从传入的 AMQP"),n("code",[e._v("Message")]),e._v("映射的头,并且有效负载是转换后的"),n("code",[e._v("body")]),e._v("。"),n("br"),e._v("当设置为"),n("code",[e._v("EXTRACT_PAYLOADS")]),e._v("时,有效负载是"),n("code",[e._v("List>")]),e._v(",其中元素是从 AMQP"),n("code",[e._v("Message")]),e._v("体转换而来的。"),n("code",[e._v("EXTRACT_PAYLOADS_WITH_HEADERS")]),e._v("类似于"),n("code",[e._v("EXTRACT_PAYLOADS")]),e._v(",但是,除此之外,每个消息的头从"),n("code",[e._v("MessageProperties")]),e._v("映射到相应索引处的"),n("code",[e._v("List