第22章. 异步和消息

第22章. 异步和消息

 

Seam 使异步执行来自网页请求的工作变得非常容易。在多数人在Java EE中考虑异步时，他们想到的是使用JMS。在Seam中，这的确是一种解决问题的方法，并且在你有严格和明确定义的服务质量需求时，这是正确的方法。 Seam利用Seam组件使发送和接收JMS消息变得非常容易。

但是对多数用例而言，用JMS就太夸张了。Seam在你的调度器（dispatchers） 选择层之上分层了一个简单的异步方法和事件机能：

• java.util.concurrent.ScheduledThreadPoolExecutor (默认)
• the EJB timer service (对 EJB 3.0 环境)
• Quartz

 

22.1. 异步

 

异步事件和方法调用具有与基本的分配器机制一样期望的相同服务质量。基于ScheduledThreadPoolExecutor的默认分配器（dispatcher）能有效地完成任务，但不提供对持久化异步任务的支持，因此不能保证一个任务总会被真正地执行。如果你工作在支持EJB3的环境，增加下列行到components.xml:

<async:timer-service-dispatcher/>

 

那么，你的异步任务会被容器的EJB定时器服务处理。如果你不熟悉定时器服务，不用担心，假如你想在Seam中使用异步方法，你不必直接与它交互。知道的重要事情是任何好的EJB3实现都会有使用持久化定时器的选项，它将保证任务最终会被处理。

另外的一个选择是使用开源Quartz库来管理异步方法。你需要捆绑Quartz库JAR (在lib目录会找到) 在你的EAR中，并application.xml声明它作为一个Java 模块。Quartz分配器可以通过增加一个Quartz 性文件到类目录被配置。它必须被命名为seam.quartz.properties。 此外，你需要增加下列行到components.xml，安装Quartz分配器。

 

<async:quartz-dispatcher/>

 

Seam API使用ScheduledThreadPoolExecutor、 EJB3定时器和Quartz调度器（Scheduler）为默认值很大程度上是相同的。他们可以通过增加一行到components.xml中实现“即插即用”

22.1.1. 异步方法

最简单的形式，一个异步调用仅让来自调用者的方法被异步执行（以不同的线程）。在我们想即时响应客户端时，我们常使用一个异步调用，并让一些费时的工作在后台处理。在使用AJAX的应用程序中这个模式运行良好，客户端可以自动轮询服务器的工作成果。

对EJB组件，我们注释本地接口来指定一个方法被异步执行。

@Local

 

public interface PaymentHandler

 

{

 

    @Asynchronous

 

    public void processPayment(Payment payment);

 

}

(对JavaBean组件 ，如果我们喜欢，我们可以注释组件实现类)

对bean类而言，异步的使用是透明的：

@Stateless

 

@Name("paymentHandler")

 

public class PaymentHandlerBean implements PaymentHandler

 

{

 

    public void processPayment(Payment payment)

 

    {

 

        //do some work!

 

    }

 

}

并对客户端也是透明的：

@Stateful

 

@Name("paymentAction")

 

public class CreatePaymentAction

 

{

 

    @In(create=true) PaymentHandler paymentHandler;

 

    @In Bill bill;

 

        public String pay()

 

    {

 

        paymentHandler.processPayment( new Payment(bill) );

 

        return "success";

 

    }

 

}

异步方法在一个全新的事件（event）上下文中被执行，并且没有访问调用者的会话（session）或对话（ conversation）上下文的状态，业务流程（business process）上下文被传播。

异步方法调用可以使用@Duration、 @Expiration和 @IntervalDuration 注释来预定计划稍后执行。

@Local

 

public interface PaymentHandler

 

{

    @Asynchronous

 

    public void processScheduledPayment(Payment payment, @Expiration Date date);

 

    @Asynchronous

 

    public void processRecurringPayment(Payment payment, @Expiration Date date,  @IntervalDuration Long interval)'

 

}

 

@Stateful

 

@Name("paymentAction")

 

public class CreatePaymentAction

 

{

    @In(create=true) PaymentHandler paymentHandler;

 

    @In Bill bill;

 

       public String schedulePayment()

 

    {

        paymentHandler.processScheduledPayment( new Payment(bill), bill.getDueDate() );

 

        return "success";

   }

 

    public String scheduleRecurringPayment()

 

    {

 

        paymentHandler.processRecurringPayment( new Payment(bill), bill.getDueDate(),  ONE_MONTH );

 

        return "success";

 

    }

 

}

客户端和服务都可以访问与调用相关联的定时器对象。显示在下面的定时器对象是EJB3定时器，在你使用EJB3分配器（dispatcher）时。对于默认的ScheduledThreadPoolExecutor，返回的对象是JDK的Future。对于Quartz 分配器（dispatcher）, 它返回 QuartzTriggerHandle，我们在下节讨论它。

@Local

 

public interface PaymentHandler

 

{

    @Asynchronous

 

    public Timer processScheduledPayment(Payment payment, @Expiration Date date);

 

}

 

@Stateless

 

@Name("paymentHandler")

 

public class PaymentHandlerBean implements PaymentHandler

 

{

 

    @In Timer timer;

 

        public Timer processScheduledPayment(Payment payment, @Expiration Date date)

 

    {

        //do some work!

 

        

 

        return timer; //注意返回的值完全被忽略

    }

 

}

 

 

@Stateful

 

@Name("paymentAction")

 

public class CreatePaymentAction

 

{

 

    @In(create=true) PaymentHandler paymentHandler;

 

    @In Bill bill;

 

    public String schedulePayment()

 

    {

 

        Timer timer = paymentHandler.processScheduledPayment( new Payment(bill), bill.getDueDate() );

 

        return "success";

 

    }

 

}

异步方法不能返回任何其它值给调用者。

22.1.2. 使用Quartz分配器的异步方法

Quartz分配器（如何安装它见前文）允许使用上文的@Asynchronous, @Duration, @Expiration, 和 @IntervalDuration注释。 然而它有一些超强的附加功能。Quartz分配器支持三个新的注释。

@FinalExpiration注释为经常性的任务指定一个结束日期。注意，你可以注入 QuartzTriggerHandle。

 

 

        @In QuartzTriggerHandle timer;

 

          //  在" processor "组件中定义方法 

 

    @Asynchronous

 

     public QuartzTriggerHandle schedulePayment(@Expiration Date when,  @IntervalDuration Long interval, @FinalExpiration Date endDate, Payment payment) 

 

    { 

        // 做重复或长期运行的任务，直到结束日期（endDate）

 

    }

    ... ...

 

    // 在业务逻辑处理代码中制定计划任务

 

    // 现在开始, 每隔一小时重复，2010年5月10号结束

 

    Calendar cal = Calendar.getInstance ();

 

    cal.set (2010, Calendar.MAY, 10);

 

    processor.schedulePayment(new Date(), 60*60*1000, cal.getTime(), payment);

注意，这个方法返回QuartzTriggerHandle对象，稍后你可以用它来停止、暂停和重新开始日程安排程序（scheduler）。QuartzTriggerHandle对象被系列化，所以，如果你需要保持它一段较长的时间，你可以存储它到数据库。

QuartzTriggerHandle handle =processor.schedulePayment(payment.getPaymentDate(),  payment.getPaymentCron(),payment);

 

        payment.setQuartzTriggerHandle( handle );

 

        // 存 payment到数据库

 

             // 稍后 ...

 

             // 从数据库取回 payment

 

        // 取消剩余的计划任务

 

        payment.getQuartzTriggerHandle().cancel();

 @IntervalCron支持任务行程安排的Unix cron工作语法。例如，下面的异步方法运行在3月的每个星期三的下午2点10分到下午2点44分。

    // 定义方法

 

    @Asynchronous

 

    public QuartzTriggerHandle schedulePayment(@Expiration Date when, @IntervalCron String cron,  Payment payment) 

 

    { 

 

        // 做重复或长期运行的任务

 

    }

    ... ...

    //在业务逻辑处理代码中制定计划任务

 

    QuartzTriggerHandle handle = processor.schedulePayment(new Date(), "0 10,44 14 ? 3 WED", payment);

@IntervalBusinessDay注释支持在“第N营业日”调用的情形。例如，下面的异步方法运行在每月的第2个营业日的14:00点。默认时，它从营业日中排除了直到2010年之前的所有周末和美国联邦假期。

    // 定义

 

    @Asynchronous

 

    public QuartzTriggerHandle schedulePayment(@Expiration Date when, @IntervalBusinessDay NthBusinessDay nth,  Payment payment) 

 

    { 

        // 做重复或长期运行的任务

 

    }

 

     ... ...

 

    // 在业务逻辑处理代码中制定计划任务

 

    QuartzTriggerHandle handle = processor.schedulePayment(new Date(),  new NthBusinessDay(2, "14:00", WEEKLY), payment);

NthBusinessDay对象包含调用触发器的配置。 你可以通过additionalHolidays属性指定更多的假日（例如，公司假日，非美国假日等等）。

 

public class NthBusinessDay implements Serializable

 

{

 

      int n;

 

      String fireAtTime;

 

      List <Date> additionalHolidays;

 

      BusinessDayIntervalType interval;

 

      boolean excludeWeekends;

 

      boolean excludeUsFederalHolidays;

 

 

      public enum BusinessDayIntervalType { WEEKLY, MONTHLY, YEARLY }

 

 

      public NthBusinessDay ()

 

      {

 

        n = 1;

 

        fireAtTime = "12:00";

 

        additionalHolidays = new ArrayList <Date> ();

 

        interval = BusinessDayIntervalType.WEEKLY;

 

        excludeWeekends = true;

 

        excludeUsFederalHolidays = true;

 

      }     

 

      ... ...

 

}

@IntervalDuration, @IntervalCron, 和 @IntervalNthBusinessDay注释是互斥的。 如果使用它们在同一个方法中，会抛出RuntimeException。

22.1.3. 异步事件

组件驱动事件也可以是异步的。为引发一个异步处理事件，简单调用Events 类的raiseAsynchronousEvent()方法就行了。 要安排一个定时事件，调用 raiseTimedEvent()方法, 传递一个schedule对象给它(对默认分配器或定时器服务分配器，使用TimerSchedule)。 组件可以用通常的方法观察异步事件，但是要记着只有业务流程（business process）上下文被传播到异步线程。

22.1.4. 根据异步调用处理异常

当一个异常通过异步分配器传播时，每种异步分配器的行为表现不同。例如，java.util.concurrent 分配器会暂停一个重复调用的进一步执行，而EJB3定时服务会取消（swallow）异常。因此，Seam 捕获异步调用产生的所有异常，在它达到分配器之前。

默认时，异步执行产生的任何异常将被捕获，并记录在错误级别。你可以自定义这种全局行为，通过覆盖org.jboss.seam.async.asynchronousExceptionHandler组件:

 

@Scope(ScopeType.STATELESS)

 

@Name("org.jboss.seam.async.asynchronousExceptionHandler")

 

public class MyAsynchronousExceptionHandler extends AsynchronousExceptionHandler { 

 

 

   @Logger Log log;

 

   @In Future timer;

 

   @Override

 

   public void handleException(Exception exception) {

 

      log.debug(exception);

 

      timer.cancel(false);

 

   }

 

 }

这里,例如， 使用java.util.concurrent 分配器，我们注入它的控制（control）对象，并且在遭遇一个异常时，取消以后所有调用。

对个别组件你也可以改变这种行为，通过在这个组件中实现方法public void handleAsynchronousException(Exception exception); 例如 ：

   public void handleAsynchronousException(Exception exception) {

 

      log.fatal(exception);

 

   }

22.2. 在Seam中的消息

Seam可轻松在Seam组件间发送和接收JMS消息。

22.2.1. 配置Configuration

为配置发送JMS消息的Seam的基础设施，你需要告诉Seam你想发送消息去的所有topics（主题）和queues（队列） ，并也要告诉Seam在什么地方找到QueueConnectionFactory（队列连接工厂）和 TopicConnectionFactory（主题连接工厂）。

Seam默认使用UIL2ConnectionFactory， 一个使用JBossMQ（一个实现了JMS 1.1 规范的JMS服务器）的通用连接工厂。如果你使用了一些其他JMS供应者，你需要在seam.properties，web.xml或components.xml中设置一个或两个queueConnection.queueConnectionFactoryJndiName 和topicConnection.topicConnectionFactoryJndiName。

你也需要在components.xml列出topics和queues，安装Seam管理的 TopicPublishers和 QueueSenders:

 

<jms:managed-topic-publisher name="stockTickerPublisher"   auto-create="true" topic-jndi-name="topic/stockTickerTopic"/>

 

 

 

<jms:managed-queue-sender name="paymentQueueSender"  auto-create="true"  queue-jndi-name="queue/paymentQueue"/>

 

22.2.2. 发送消息

现在，你可以注入一个JMS TopicPublisher 和TopicSession到任何组件：

@In 

 

private TopicPublisher stockTickerPublisher;   

 

@In 

 

private TopicSession topicSession;

 

 

public void publish(StockPrice price) {

 

      try

 

      {

 

         stockTickerPublisher.publish( topicSession.createObjectMessage(price) );

 

      } 

 

      catch (Exception ex)

 

      {

 

         throw new RuntimeException(ex);

 

      } 

 

}

或者, 使用队列:

@In

 

private QueueSender paymentQueueSender;   

 

@In

 

private QueueSession queueSession;

 

public void publish(Payment payment) {

 

      try

 

      {

 

         paymentQueueSender.send( queueSession.createObjectMessage(payment) );

 

      } 

 

      catch (Exception ex)

 

      {

 

         throw new RuntimeException(ex);

 

      } 

 

}

22.2.3. 使用消息驱动bean接收消息

你也可能使用任何EJB3消息驱动bean处理消息。消息驱动bean甚至可以是Seam组件，在这种情况下，注入其它事件和应用程序作用域的Seam组件成为可能。

22.2.4. 在客户端接收消息

Seam远程让你根据客户边JavaScript订阅一个JMS主题。这在第25章 远程描述。