(window.webpackJsonp=window.webpackJsonp||[]).push([[502],{930:function(e,a,n){"use strict";n.r(a);var r=n(56),o=Object(r.a)({},(function(){var e=this,a=e.$createElement,n=e._self._c||a;return n("ContentSlotsDistributor",{attrs:{"slot-key":e.$parent.slotKey}},[n("h1",{attrs:{id:"事务支持"}},[n("a",{staticClass:"header-anchor",attrs:{href:"#事务支持"}},[e._v("#")]),e._v(" 事务支持")]),e._v(" "),n("h2",{attrs:{id:"事务支持-2"}},[n("a",{staticClass:"header-anchor",attrs:{href:"#事务支持-2"}},[e._v("#")]),e._v(" 事务支持")]),e._v(" "),n("p",[e._v("本章介绍 Spring 集成对事务的支持。它涵盖以下主题：")]),e._v(" "),n("ul",[n("li",[n("p",[n("a",{attrs:{href:"#understanding-transaction"}},[e._v("理解消息流中的事务")])])]),e._v(" "),n("li",[n("p",[n("a",{attrs:{href:"#transaction-boundaries"}},[e._v("交易边界")])])]),e._v(" "),n("li",[n("p",[n("a",{attrs:{href:"#transaction-synchronization"}},[e._v("事务同步")])])]),e._v(" "),n("li",[n("p",[n("a",{attrs:{href:"#pseudo-transactions"}},[e._v("伪交易")])])])]),e._v(" "),n("h3",{attrs:{id:"理解消息流中的事务"}},[n("a",{staticClass:"header-anchor",attrs:{href:"#理解消息流中的事务"}},[e._v("#")]),e._v(" 理解消息流中的事务")]),e._v(" "),n("p",[e._v("Spring 集成公开了几个钩子，以满足消息流的事务需求。为了更好地理解这些钩子以及如何从中受益，我们必须首先重新审视可以用来启动消息流的六种机制，并了解如何在这些机制中的每一种中解决这些流的事务需求。")]),e._v(" "),n("p",[e._v("以下六种机制启动消息流（本手册提供了每种机制的详细信息）：")]),e._v(" "),n("ul",[n("li",[n("p",[e._v("Gateway Proxy:一种基本的消息传递网关。")])]),e._v(" "),n("li",[n("p",[e._v("消息通道：与"),n("code",[e._v("MessageChannel")]),e._v("方法直接交互（例如，"),n("code",[e._v("channel.send(message)")]),e._v("）。")])]),e._v(" "),n("li",[n("p",[e._v("Message Publisher:在 Spring bean 上作为方法调用的副产品发起消息流的方法。")])]),e._v(" "),n("li",[n("p",[e._v("入站通道适配器和网关：发起消息流的方式基于将第三方系统与 Spring 集成消息传递系统的连接（例如，"),n("code",[e._v("[JmsMessage] → Jms Inbound Adapter[SI Message] → SI Channel")]),e._v("）。")])]),e._v(" "),n("li",[n("p",[e._v("调度器：根据预先配置的调度器分发的调度事件发起消息流的方法。")])]),e._v(" "),n("li",[n("p",[e._v("Poller:与调度器类似，这是一种基于调度或由预先配置的 Poller 分发的基于间隔的事件发起消息流的方法。")])])]),e._v(" "),n("p",[e._v("我们可以将这六种机制分为两大类：")]),e._v(" "),n("ul",[n("li",[n("p",[e._v("由用户进程发起的消息流：此类别中的示例场景将调用网关方法或显式地将"),n("code",[e._v("Message")]),e._v("发送到"),n("code",[e._v("MessageChannel")]),e._v("。换句话说，这些消息流依赖于要发起的第三方进程（例如你编写的某些代码）。")])]),e._v(" "),n("li",[n("p",[e._v("由守护进程启动的消息流：此类别中的示例场景包括一个 Poller 轮询消息队列以启动带有轮询消息的新消息流，或者一个调度器通过创建新消息并在预定义的时间启动消息流来调度该进程。")])])]),e._v(" "),n("p",[e._v("显然，网关代理、"),n("code",[e._v("MessageChannel.send(…​)")]),e._v("和"),n("code",[e._v("MessagePublisher")]),e._v("都属于第一类，入站适配器和网关、调度器和 poller 属于第二类。")]),e._v(" "),n("p",[e._v("那么，如何在每个类别中的不同场景中解决事务需求，是否需要 Spring 集成来为特定场景提供有关事务的明确内容？或者，你可以使用 Spring 的事务支持吗？")]),e._v(" "),n("p",[e._v("Spring 其本身为事务管理提供了一流的支持。因此，我们在这里的目标不是提供新的东西，而是使用 Spring 来受益于其现有的事务支持。换句话说，作为一个框架，我们必须将钩子公开到 Spring 的事务管理功能。然而，由于 Spring 集成配置是基于 Spring 配置的，因此我们不需要总是公开这些钩子，因为 Spring 已经公开了它们。毕竟，每个 Spring 集成组件都是 Spring  Bean。")]),e._v(" "),n("p",[e._v("考虑到这个目标，我们可以再次考虑两个场景：由用户进程发起的消息流和由守护进程发起的消息流。")]),e._v(" "),n("p",[e._v("由用户进程发起并在 Spring 应用程序上下文中配置的消息流受到此类进程的通常事务配置的约束。因此，不需要通过 Spring 集成显式地配置它们来支持事务。交易可以也应该通过 Spring 的标准交易支持来启动。 Spring 集成消息流自然尊重组件的事务语义，因为它本身由 Spring 配置。例如，网关或服务激活器方法可以用"),n("code",[e._v("@Transactional")]),e._v("进行注释，或者可以在 XML 配置中定义"),n("code",[e._v("TransactionInterceptor")]),e._v("，并使用切入点表达式指向应该是事务性的特定方法。最重要的是，在这些场景中，你可以完全控制事务配置和边界。")]),e._v(" "),n("p",[e._v("然而，当涉及到由守护进程发起的消息流时，情况就有些不同了。尽管由开发人员进行了配置，但这些流并不直接涉及要启动的人工或其他进程。这些是基于触发器的流，由触发器进程（守护进程）基于进程的配置发起。例如，我们可以让调度器在每个星期五晚上发起消息流。我们还可以配置一个触发器，该触发器每秒钟启动一次消息流，以此类推。因此，我们需要一种方法，让这些基于触发器的流程知道我们的意图，即使生成的消息流是事务性的，以便每当启动新的消息流时都可以创建事务上下文。换句话说，我们需要公开一些事务配置，但仅限于委托给 Spring 已经提供的事务支持（就像我们在其他场景中所做的那样）。")]),e._v(" "),n("h4",{attrs:{id:"poller-事务支持"}},[n("a",{staticClass:"header-anchor",attrs:{href:"#poller-事务支持"}},[e._v("#")]),e._v(" Poller 事务支持")]),e._v(" "),n("p",[e._v("Spring 集成为 Pollers 提供了事务支持。Poller 是一种特殊类型的组件，因为在 Poller 任务中，我们可以针对本身是事务性的资源调用"),n("code",[e._v("receive()")]),e._v("，从而在事务的边界中包括"),n("code",[e._v("receive()")]),e._v("调用，这使它在任务失败的情况下被回滚。如果我们要为通道添加相同的支持，那么添加的事务将影响从"),n("code",[e._v("send()")]),e._v("调用开始的所有下游组件。这为事务划分提供了相当广泛的范围，而没有任何强有力的理由，特别是当 Spring 已经提供了几种方法来解决下游任何组件的事务需求时。然而，将"),n("code",[e._v("receive()")]),e._v("方法包含在事务边界中，是 Pollers 的“有力理由”。")]),e._v(" "),n("p",[e._v("在配置 Poller 的任何时候，都可以通过使用"),n("code",[e._v("transactional")]),e._v("子元素及其属性来提供事务配置，如下例所示：")]),e._v(" "),n("div",{staticClass:"language- extra-class"},[n("pre",{pre:!0,attrs:{class:"language-text"}},[n("code",[e._v('<int:poller max-messages-per-poll="1" fixed-rate="1000">\n    <transactional transaction-manager="txManager"\n                   isolation="DEFAULT"\n                   propagation="REQUIRED"\n                   read-only="true"\n                   timeout="1000"/>\n</poller>\n')])])]),n("p",[e._v("前面的配置看起来类似于本机 Spring 事务配置。你仍然必须提供对事务管理器的引用，并且要么指定事务属性，要么依赖默认值（例如，如果没有指定“transactionmanager”属性，则默认使用 Bean 命名的“transactionmanager”）。在内部，该过程被包装在 Spring 的本机事务中，其中"),n("code",[e._v("TransactionInterceptor")]),e._v("负责处理事务。有关如何配置事务管理器、事务管理器的类型（例如 JTA、DataSource 和其他）以及与事务配置有关的其他详细信息，请参见"),n("a",{attrs:{href:"https://docs.spring.io/spring/docs/current/spring-framework-reference/data-access.html#transaction",target:"_blank",rel:"noopener noreferrer"}},[e._v("Spring Framework Reference Guide"),n("OutboundLink")],1),e._v("。")]),e._v(" "),n("p",[e._v("在前面的配置中，由这个 Poller 发起的所有消息流都是事务性的。有关 Poller 事务配置的更多信息和详细信息，请参见"),n("RouterLink",{attrs:{to:"/spring-integration/jdbc.html#jdbc-polling-transactions"}},[e._v("轮询和交易")]),e._v("。")],1),e._v(" "),n("p",[e._v("在运行 Poller 时，除了事务之外，你可能还需要解决几个更多的交叉问题。为了帮助实现这一点，Poller 元素接受一个"),n("code",[e._v("<advice-chain>")]),e._v("子元素，它允许你定义一个定制的建议实例链，以应用于 Poller。（有关更多详细信息，请参见"),n("RouterLink",{attrs:{to:"/spring-integration/polling-consumer.html#pollable-message-source"}},[e._v("可选消息源")]),e._v("。）在 Spring Integration2.0 中，Poller 进行了重构工作，现在使用代理机制来解决事务性问题以及其他跨领域问题。这项工作产生的一个重大变化是，我们使"),n("code",[e._v("<transactional>")]),e._v("和"),n("code",[e._v("<advice-chain>")]),e._v("元素是互斥的。这背后的理由是，如果你需要一个以上的建议，其中一个是事务建议，那么你可以将其包含在"),n("code",[e._v("<advice-chain>")]),e._v("中，这样做与以前一样方便，但具有更多的控制权，因为你现在可以选择将建议按所需的顺序放置。下面的示例展示了如何做到这一点：")],1),e._v(" "),n("div",{staticClass:"language- extra-class"},[n("pre",{pre:!0,attrs:{class:"language-text"}},[n("code",[e._v('<int:poller max-messages-per-poll="1" fixed-rate="10000">\n  <advice-chain>\n    <ref bean="txAdvice"/>\n    <ref bean="someOtherAdviceBean" />\n    <beans:bean class="foo.bar.SampleAdvice"/>\n  </advice-chain>\n</poller>\n\n<tx:advice id="txAdvice" transaction-manager="txManager">\n  <tx:attributes>\n    <tx:method name="get*" read-only="true"/>\n    <tx:method name="*"/>\n  </tx:attributes>\n</tx:advice>\n')])])]),n("p",[e._v("前面的示例显示了 Spring 事务通知（"),n("code",[e._v("txAdvice")]),e._v("）的基本基于 XML 的配置，并将其包含在由 Poller 定义的"),n("code",[e._v("<advice-chain>")]),e._v("中。如果只需要解决 Poller 的事务问题，仍然可以使用"),n("code",[e._v("<transactional>")]),e._v("元素作为方便。")]),e._v(" "),n("h3",{attrs:{id:"事务边界"}},[n("a",{staticClass:"header-anchor",attrs:{href:"#事务边界"}},[e._v("#")]),e._v(" 事务边界")]),e._v(" "),n("p",[e._v("另一个重要因素是消息流中事务的边界。当事务启动时，事务上下文被绑定到当前线程。因此，无论消息流中有多少个端点和通道，只要确保消息流在同一线程上继续，事务上下文都将被保留。一旦你通过引入"),n("em",[e._v("可选频道")]),e._v("或"),n("em",[e._v("执行器通道")]),e._v("来打破它，或者在某些服务中手动启动一个新线程，事务边界也将被打破。从本质上讲，事务将在那里结束，如果在线程之间进行了成功的切换，那么流将被认为是成功的，并且将发送一个提交信号，即使流将继续，并且仍然可能在下游某个地方导致异常。如果这样的流是同步的，那么该异常将被抛回消息流的发起者，该发起者也是事务上下文的发起者，事务将导致回滚。中间立场是在线程边界被打破的任何一点使用事务通道。例如，你可以使用一个队列支持的通道，该通道将委托给事务性 MessageStore 策略，或者你可以使用一个 JMS 支持的通道。")]),e._v(" "),n("h3",{attrs:{id:"事务同步"}},[n("a",{staticClass:"header-anchor",attrs:{href:"#事务同步"}},[e._v("#")]),e._v(" 事务同步")]),e._v(" "),n("p",[e._v("在某些环境中，它有助于将操作与包含整个流程的事务同步。例如，考虑在执行多个数据库更新的流的开始处使用"),n("code",[e._v("<file:inbound-channel-adapter/>")]),e._v("。如果事务提交，我们可能希望将文件移动到"),n("code",[e._v("success")]),e._v("目录，而如果事务回滚，我们可能希望将文件移动到"),n("code",[e._v("failure")]),e._v("目录。")]),e._v(" "),n("p",[e._v("Spring Integration2.2 引入了将这些操作与事务同步的能力。此外，如果你没有“实际”事务，但仍然希望在成功或失败时执行不同的操作，则可以配置"),n("code",[e._v("PseudoTransactionManager")]),e._v("。有关更多信息，请参见"),n("a",{attrs:{href:"#pseudo-transactions"}},[e._v("伪交易")]),e._v("。")]),e._v(" "),n("p",[e._v("下面的清单显示了该特性的关键策略接口：")]),e._v(" "),n("div",{staticClass:"language- extra-class"},[n("pre",{pre:!0,attrs:{class:"language-text"}},[n("code",[e._v("public interface TransactionSynchronizationFactory {\n\n    TransactionSynchronization create(Object key);\n}\n\npublic interface TransactionSynchronizationProcessor {\n\n    void processBeforeCommit(IntegrationResourceHolder holder);\n\n    void processAfterCommit(IntegrationResourceHolder holder);\n\n    void processAfterRollback(IntegrationResourceHolder holder);\n\n}\n")])])]),n("p",[e._v("工厂负责创建一个["),n("code",[e._v("TransactionSynchronization")]),e._v("]（https://DOCS. Spring.io/ Spring-framework/DOCS/current/javadoc-api/org/springframework/transaction/support/transactionsynchronization.html）对象。你可以实现自己的或使用框架提供的："),n("code",[e._v("DefaultTransactionSynchronizationFactory")]),e._v("。此实现返回一个"),n("code",[e._v("TransactionSynchronization")]),e._v("，该实现将委托给默认的"),n("code",[e._v("TransactionSynchronizationProcessor")]),e._v("实现："),n("code",[e._v("ExpressionEvaluatingTransactionSynchronizationProcessor")]),e._v("。此处理器支持三个 SPEL 表达式："),n("code",[e._v("beforeCommitExpression")]),e._v("、"),n("code",[e._v("afterCommitExpression")]),e._v("和"),n("code",[e._v("afterRollbackExpression")]),e._v("。")]),e._v(" "),n("p",[e._v("对于那些熟悉事务的人来说，这些操作应该是不言自明的。在每种情况下，"),n("code",[e._v("#root")]),e._v("变量都是原始的"),n("code",[e._v("Message")]),e._v("。在某些情况下，其他 SPEL 变量是可用的，这取决于由 poller 进行民意测验的"),n("code",[e._v("MessageSource")]),e._v("。例如，"),n("code",[e._v("MongoDbMessageSource")]),e._v("提供了"),n("code",[e._v("#mongoTemplate")]),e._v("变量，该变量引用消息源的"),n("code",[e._v("MongoTemplate")]),e._v("。类似地，"),n("code",[e._v("RedisStoreMessageSource")]),e._v("提供了"),n("code",[e._v("#store")]),e._v("变量，该变量引用了轮询创建的"),n("code",[e._v("RedisStore")]),e._v("。")]),e._v(" "),n("p",[e._v("要为特定的 Poller 启用该特性，你可以使用"),n("code",[e._v("synchronization-factory")]),e._v("属性，在 Poller 的"),n("code",[e._v("TransactionSynchronizationFactory")]),e._v("元素上提供对"),n("code",[e._v("TransactionSynchronizationFactory")]),e._v("的引用。")]),e._v(" "),n("p",[e._v("从版本 5.0 开始， Spring Integration 提供，当没有配置，但在建议链中存在类型的建议时，默认情况下将其应用于轮询端点。当使用任何开箱即用的"),n("code",[e._v("TransactionSynchronizationFactory")]),e._v("实现时，轮询端点将轮询消息绑定到当前事务上下文，并在"),n("code",[e._v("MessagingException")]),e._v("中将其作为"),n("code",[e._v("failedMessage")]),e._v("提供，如果在事务通知之后抛出了异常。当使用不实现"),n("code",[e._v("TransactionInterceptor")]),e._v("的自定义事务建议时，可以显式地配置"),n("code",[e._v("PassThroughTransactionSynchronizationFactory")]),e._v("以实现此行为。在这两种情况下，"),n("code",[e._v("MessagingException")]),e._v("都成为发送到"),n("code",[e._v("ErrorMessage")]),e._v("的"),n("code",[e._v("errorChannel")]),e._v("的有效负载，而原因是建议引发的原始异常。在此之前，"),n("code",[e._v("ErrorMessage")]),e._v("的有效负载是由建议引发的原始异常，并且不提供对"),n("code",[e._v("failedMessage")]),e._v("信息的引用，这使得很难确定事务提交问题的原因。")]),e._v(" "),n("p",[e._v("Spring 为了简化这些组件的配置，集成为默认工厂提供了名称空间支持。下面的示例展示了如何使用名称空间来配置文件入站通道适配器：")]),e._v(" "),n("div",{staticClass:"language- extra-class"},[n("pre",{pre:!0,attrs:{class:"language-text"}},[n("code",[e._v('<int-file:inbound-channel-adapter id="inputDirPoller"\n    channel="someChannel"\n    directory="/foo/bar"\n    filter="filter"\n    comparator="testComparator">\n    <int:poller fixed-rate="5000">\n        <int:transactional transaction-manager="transactionManager" synchronization-factory="syncFactory" />\n    </int:poller>\n</int-file:inbound-channel-adapter>\n\n<int:transaction-synchronization-factory id="syncFactory">\n    <int:after-commit expression="payload.renameTo(new java.io.File(\'/success/\' + payload.name))"\n        channel="committedChannel" />\n    <int:after-rollback expression="payload.renameTo(new java.io.File(\'/failed/\' + payload.name))"\n        channel="rolledBackChannel" />\n</int:transaction-synchronization-factory>\n')])])]),n("p",[e._v("SPEL 评估的结果作为有效负载发送到"),n("code",[e._v("committedChannel")]),e._v("或"),n("code",[e._v("rolledBackChannel")]),e._v("（在这种情况下，这将是"),n("code",[e._v("Boolean.TRUE")]),e._v("或"),n("code",[e._v("Boolean.FALSE")]),e._v("——"),n("code",[e._v("java.io.File.renameTo()")]),e._v("方法调用的结果）。")]),e._v(" "),n("p",[e._v("如果你希望将整个有效载荷发送给进一步的 Spring 集成处理，请使用’payload’表达式。")]),e._v(" "),n("table",[n("thead",[n("tr",[n("th"),e._v(" "),n("th",[e._v("重要的是要理解这会使操作与事务同步，"),n("br"),e._v("它不会使本质上不是事务性的资源实际上是事务性的。，相反，"),n("br"),e._v("，事务（无论是 JDBC 还是其他方式）在轮询之前启动，并在流完成时提交或回滚，然后是同步操作。，如果提供自定义"),n("br"),n("br"),e._v("，它负责创建一个资源同步，该同步使绑定的资源在事务完成时自动解除绑定。"),n("br"),e._v("默认值"),n("code",[e._v("TransactionSynchronizationFactory")]),e._v("是通过返回一个"),n("code",[e._v("ResourceHolderSynchronization")]),e._v("的子类来实现的，默认值"),n("code",[e._v("shouldUnbindAtCompletion()")]),e._v("返回"),n("code",[e._v("true")]),e._v("。")])])]),e._v(" "),n("tbody")]),e._v(" "),n("p",[e._v("除了"),n("code",[e._v("after-commit")]),e._v("和"),n("code",[e._v("after-rollback")]),e._v("表达式外，还支持"),n("code",[e._v("before-commit")]),e._v("表达式。在这种情况下，如果评估（或下游处理）抛出异常，事务将被回滚而不是提交。")]),e._v(" "),n("h3",{attrs:{id:"伪事务"}},[n("a",{staticClass:"header-anchor",attrs:{href:"#伪事务"}},[e._v("#")]),e._v(" 伪事务")]),e._v(" "),n("p",[e._v("在阅读了"),n("a",{attrs:{href:"#transaction-synchronization"}},[e._v("事务同步")]),e._v("部分之后，你可能会认为在流完成时执行这些“成功”或“失败”操作是有用的，即使在 Poller 的下游没有“真正的”事务资源（例如 JDBC）。例如，考虑一个“<file:inbound-channel-adapter/>”后跟一个“<ftp:outbout-channel-adapter/>”。这两个组件都不是事务性的，但是我们可能希望根据 FTP 传输的成功或失败，将输入文件移动到不同的目录。")]),e._v(" "),n("p",[e._v("为了提供这种功能，框架提供了"),n("code",[e._v("PseudoTransactionManager")]),e._v("，即使在不涉及真正的事务资源的情况下，也能够实现上述配置。如果流正常完成，则调用"),n("code",[e._v("beforeCommit")]),e._v("和"),n("code",[e._v("afterCommit")]),e._v("同步。一旦失败，将调用"),n("code",[e._v("afterRollback")]),e._v("同步。因为它不是一个真正的事务，所以不会发生实际的提交或回滚。伪事务是一种用来实现同步功能的载体。")]),e._v(" "),n("p",[e._v("要使用"),n("code",[e._v("PseudoTransactionManager")]),e._v("，可以将其定义为 <bean/>，就像配置实际事务管理器一样。下面的示例展示了如何做到这一点：")]),e._v(" "),n("div",{staticClass:"language- extra-class"},[n("pre",{pre:!0,attrs:{class:"language-text"}},[n("code",[e._v('<bean id="transactionManager" class="o.s.i.transaction.PseudoTransactionManager" />\n')])])]),n("h3",{attrs:{id:"无反应事务"}},[n("a",{staticClass:"header-anchor",attrs:{href:"#无反应事务"}},[e._v("#")]),e._v(" 无反应事务")]),e._v(" "),n("p",[e._v("从版本 5.3 开始，对于返回反应类型的端点，"),n("code",[e._v("ReactiveTransactionManager")]),e._v("通知也可以与"),n("code",[e._v("TransactionInterceptor")]),e._v("通知一起使用。这包括"),n("code",[e._v("MessageSource")]),e._v("和"),n("code",[e._v("ReactiveMessageHandler")]),e._v("实现（例如"),n("code",[e._v("ReactiveMongoDbMessageSource")]),e._v("），它们产生带有"),n("code",[e._v("Flux")]),e._v("或"),n("code",[e._v("Mono")]),e._v("有效载荷的消息。当它们的应答负载也是某种反应性类型时，所有其他产生应答的消息处理程序实现都可以依赖于"),n("code",[e._v("ReactiveTransactionManager")]),e._v("。")])])}),[],!1,null,null,null);a.default=o.exports}}]);