Spring认证中国教育管理中心-Apache Geode 的 Spring 数据教程十九

原创

IT胶囊

发布于 2021-12-29 16:12:13

9110

发布于 2021-12-29 16:12:13

文章被收录于专栏：IT技能应用IT技能应用

原标题：Spring认证中国教育管理中心-Apache Geode 的 Spring 数据教程十九（Spring中国教育管理中心）

7.5.使用@TransactionalEventListener

使用事务时，可能需要注册一个侦听器，以便在事务提交之前或之后或发生回滚之后执行某些操作。

Spring Data for Apache Geode 使创建侦听器变得容易，这些侦听器将在具有@ TransactionalEventListener注释的事务的特定阶段被调用。带注释的方法@TransactionalEventListener（如下所示）将在指定的被通知从事务方法发布的事件的，phase。

事务提交后事件侦听器

@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
public void handleAfterCommit(MyEvent event) {
    // do something after transaction is committed
}

为了调用上述方法，您必须从事务中发布一个事件，如下所示：

发布交易事件

@Service
class MyTransactionalService {

  @Autowired
  private final ApplicationEventPublisher applicationEventPublisher;

  @Transactional
  public <Return-Type> someTransactionalServiceMethod() {

    // Perform business logic interacting with and accessing multiple transactional resources atomically, then...

    applicationEventPublisher.publishEvent(new MyApplicationEvent(...));
  }

  ...
}

该@ TransactionalEventListener注解允许你指定交易phase在事件处理方法将被调用。选项包括：AFTER_COMMIT，AFTER_COMPLETION，AFTER_ROLLBACK，和BEFORE_COMMIT。如果未指定，则phase默认为AFTER_COMMIT。如果您希望即使不存在事务也能调用侦听器，您可以设置fallbackExecution为true。

7.6.自动交易事件发布

从 Spring Data for Apache Geode 开始Neumann/2.3，现在可以启用自动事务事件发布。

使用@ EnableGemfireCacheTransactions注释，将enableAutoTransactionEventPublishing属性设置为true。默认值为false。

启用自动交易事件发布

@EnableGemfireCacheTransactions(enableAutoTransactionEventPublishing = true)
class GeodeConfiguration { ... }

然后，您可以创建带@ TransactionalEventListener注释的 POJO 方法来处理事务阶段AFTER_COMMIT或AFTER_ROLLBACK事务阶段的事务事件。

@Component
class TransactionEventListeners {

	@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
	public void handleAfterCommit(TransactionApplicationEvent event) {
		...
	}

	@TransactionalEventListener(phase = TransactionPhase.AFTER_ROLLBACK)
	public void handleAfterRollback(TransactionApplicationEvent event) {
		...
	}
}

仅支持 TransactionPhase.AFTER_COMMIT和TransactionPhase.AFTER_ROLLBACK。 TransactionPhase.BEFORE_COMMIT不支持，因为 1) SDG 适配了 Apache GeodeTransactionListener 和TransactionWriter接口来实现自动事务事件发布，以及 2) 当 Apache Geode TransactionWriter.beforeCommit(:TransactionEvent)被调用时，它已经在AbstractPlatformTransactionManager.triggerBeforeCommit(:TransactionStatus)调用之后， @TranactionalEventListener 在事务生命周期中调用带注释的 POJO 方法。

使用自动事务事件发布，您无需 applicationEventPublisher.publishEvent(..)在应用程序@Transactional @Service方法中显式调用该方法。

但是，如果您仍然希望在“提交之前”接收事务事件，那么您仍然必须 applicationEventPublisher.publishEvent(..)在您的应用程序@Transactional @Service方法中调用该方法。

7.7.连续查询 (CQ)

Apache Geode 提供的一项强大功能是连续查询（或 CQ）。

简而言之，CQ 允许开发人员创建和注册 OQL 查询，然后在添加到 Apache Geode 的新数据与查询谓词匹配时自动收到通知。Spring Data for Apache Geode 通过 org.springframework.data.gemfire.listener包及其侦听器容器为 CQ 提供专门的支持；在功能和命名上与Spring Framework 中的 JMS 集成非常相似；事实上，熟悉 Spring 中 JMS 支持的用户应该会有宾至如归的感觉。

基本上，Apache Geode 的 Spring Data 允许 POJO 上的方法成为 CQ 的端点。只需定义查询并指示应调用的方法，以便在匹配时收到通知。Apache Geode 的 Spring Data 负责其余的工作。这与 Java EE 的消息驱动 bean 风格非常相似，但对基类或接口实现没有任何要求，基于 Apache Geode。

目前，仅在 Apache Geode 的客户端/服务器拓扑中支持连续查询。此外，使用的客户端池需要启用订阅。有关更多信息，请参阅 Apache Geode 文档。

7.7.1.连续查询侦听器容器

Spring Data for Apache Geode 通过使用 SDG 来处理 CQ 周围的基础设施，简化了 CQ 事件的创建、注册、生命周期和分派， SDGContinuousQueryListenerContainer代表用户完成了所有繁重的工作。熟悉 EJB 和 JMS 的用户应该会发现熟悉的概念，因为它的设计尽可能接近Spring Framework及其消息驱动的 POJO (MDP) 中提供的支持。

SDGContinuousQueryListenerContainer充当事件（或消息）侦听器容器；它用于从注册的 CQ 接收事件并调用注入其中的 POJO。侦听器容器负责消息接收的所有线程并分派到侦听器中进行处理。它充当 EDP（事件驱动的 POJO）和事件提供者之间的中介，负责 CQ 的创建和注册（接收事件）、资源获取和释放、异常转换等。这允许您作为应用程序开发人员编写与接收事件（并对其做出反应）相关的（可能很复杂）业务逻辑，并将样板 Apache Geode 基础设施问题委托给框架。

侦听器容器是完全可定制的。开发人员可以选择使用 CQ 线程来执行分派（同步交付）或通过定义合适的 java.util.concurrent.Executor（或 Spring 的TaskExecutor）的异步方法的新线程（来自现有池）。根据负载、侦听器的数量或运行时环境，开发人员应该更改或调整执行器以更好地满足她的需求。特别是在托管环境（例如应用服务器）中，强烈建议选择一个合适的TaskExecutor 来利用其运行时。

7.7.2.在ContinuousQueryListener与ContinuousQueryListenerAdapter

该 ContinuousQueryListenerAdapter班是Spring数据为Apache的Geode CQ支持的最后一个组件。简而言之，类允许您将几乎所有实现类公开为具有最少约束的 EDP。 ContinuousQueryListenerAdapter实现ContinuousQueryListener接口，一个简单的监听器接口，类似于 Apache Geode 的CqListener。

考虑以下接口定义。注意各种事件处理方法及其参数：

public interface EventDelegate {
     void handleEvent(CqEvent event);
     void handleEvent(Operation baseOp);
     void handleEvent(Object key);
     void handleEvent(Object key, Object newValue);
     void handleEvent(Throwable throwable);
     void handleQuery(CqQuery cq);
     void handleEvent(CqEvent event, Operation baseOp, byte[] deltaValue);
     void handleEvent(CqEvent event, Operation baseOp, Operation queryOp, Object key, Object newValue);
}

package example;

class DefaultEventDelegate implements EventDelegate {
    // implementation elided for clarity...
}

特别要注意EventDelegate接口的上述实现完全没有Apache Geode 依赖项。它确实是一个 POJO，我们可以并且将通过以下配置将其制成 EDP。

该类不必实现接口；一个接口只是用来更好地展示合约和实现之间的解耦。

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:gfe="https://www.springframework.org/schema/geode"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="
    http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
    https://www.springframework.org/schema/geode https://www.springframework.org/schema/geode/spring-geode.xsd
">

	<gfe:client-cache/>

	<gfe:pool subscription-enabled="true">
	   <gfe:server host="localhost" port="40404"/>
	</gfe:pool>

	<gfe:cq-listener-container>
	   <!-- default handle method -->
	   <gfe:listener ref="listener" query="SELECT * FROM /SomeRegion"/>
	   <gfe:listener ref="another-listener" query="SELECT * FROM /AnotherRegion" name="myQuery" method="handleQuery"/>
	</gfe:cq-listener-container>

	<bean id="listener" class="example.DefaultMessageDelegate"/>
	<bean id="another-listener" class="example.DefaultMessageDelegate"/>
  ...
<beans>

上面的例子展示了听众可以拥有的几种不同的形式；至少，需要侦听器引用和实际查询定义。但是，可以为生成的连续查询指定一个名称（用于监视）以及方法的名称（默认为handleEvent）。指定的方法可以有各种参数类型，EventDelegate接口列出了允许的类型。

上面的示例使用 Spring Data for Apache Geode 命名空间来声明事件侦听器容器并自动注册侦听器。完整的bean定义如下所示：

<!-- this is the Event Driven POJO (MDP) -->
<bean id="eventListener" class="org.springframework.data.gemfire.listener.adapter.ContinuousQueryListenerAdapter">
    <constructor-arg>
        <bean class="gemfireexample.DefaultEventDelegate"/>
    </constructor-arg>
</bean>

<!-- and this is the event listener container... -->
<bean id="gemfireListenerContainer" class="org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer">
    <property name="cache" ref="gemfireCache"/>
    <property name="queryListeners">
      <!-- set of CQ listeners -->
      <set>
        <bean class="org.springframework.data.gemfire.listener.ContinuousQueryDefinition" >
               <constructor-arg value="SELECT * FROM /SomeRegion" />
               <constructor-arg ref="eventListener"/>
        </bean>
      </set>
    </property>
</bean>