Jakarta EE 10 - Jakarta Concurrency 3.0 変更内容まとめ

はじめに
アノテーションによるリソース定義
CDI ビーンのメソッド非同期実行
並列ストリーム操作
Trigger によるスケジュール実行の強化
その他の変更

blog1.mammb.com

はじめに

Jakarta EE 10 で Jakarta Concurrency は 2.0 から 3.0 へバージョンアップします。

アノテーションによるリソース定義や Cron 形式のトリガ実行、CDIビーンの非同期実行、MicroProfile Context Propagation からの仕様取り込みなど、それなりに変更が入っています。

それぞれ見ていきましょう。

アノテーションによるリソース定義

アノテーションにて Executor などのリソース定義が可能になりました。

データソースやJMSリソースなどについては、以下参考のように以前から可能でした。

blog1.mammb.com

今回 Concurrency 系についてもリソースのアノテーションによる定義が追加された形となります。

以下のアノテーションが追加されました。

@ContextServiceDefinition
@ManagedExecutorDefinition
@ManagedScheduledExecutorDefinition
@ManagedThreadFactoryDefinition

ManagedExecutorService の定義例は以下のようになります。

@ManagedExecutorDefinition(
    name = "java:module/concurrent/MyExecutor",
    context = "java:module/concurrent/MyExecutorContext",
    hungTaskThreshold = 120000,
    maxAsync = 5)
@ContextServiceDefinition(
    name = "java:module/concurrent/MyExecutorContext",
    propagated = { SECURITY, APPLICATION })
public class MyServlet extends HttpServlet {

    @Resource(lookup = "java:module/concurrent/MyExecutor",
              name = "java:module/concurrent/env/MyExecutorRef")
    ManagedExecutorService myExecutor;
}

データソースやJMSリソースなどと同様に、デプロイメントディスクリプタのエントリとして指定された場合には、デプロイメントディスクリプタの定義が優先されます。

CDI ビーンのメソッド非同期実行

jakarta.enterprise.concurrent.Asynchronous アノテーションが新しく追加され、CDI ビーンのメソッドを非同期で実行できるようになりました。

Asynchronous アノテーションの定義は以下のようになっています。

@Documented
@Inherited
@InterceptorBinding
@Retention(RetentionPolicy.RUNTIME)
@Target({ ElementType.METHOD, ElementType.TYPE })
public @interface Asynchronous {

  @Nonbinding
  String executor() default "java:comp/DefaultManagedExecutorService";

  public static class Result {
    private static final ThreadLocal<CompletableFuture<?>> FUTURES = new ThreadLocal<CompletableFuture<?>>();
    public static <T> CompletableFuture<T> complete(final T result) { ... }
    public static <T> CompletableFuture<T> getFuture() { ... }
    public static <T> void setFuture(final CompletableFuture<T> future) { ...  }
  }
}

"java:comp/DefaultManagedExecutorService" とある様に、背後で ExecutorService により実行されます。

このアノテーションは、EJBには付与することはできず、また、MicroProfile の org.eclipse.microprofile.faulttolerance.Asynchronous アノテーションと共存することもできません。

利用例は以下のようになります。

public class ProductRecommendations {

   @Asynchronous
   public CompletableFuture<Set<Item>> findSimilar(Cart cart, History h) {
      Set<Item> combined = new LinkedHashSet<Item>();
      // ...
      return Asynchronous.Result.complete(combined);
   }
}

呼び出し側は CompletionStage.thenAccept() などで受けることができます。

public class CheckoutServlet extends HttpServlet {

  @Inject
  ProductRecommendations recommendations;

  public void doGet(HttpServletRequest req, HttpServletResponse resp)
      throws ServletException {
    // ...
    recommendations.findSimilar(cust.getCart(), cust.getHistory())
        .thenAccept(recommended -> { ... });
   }
}

トランザクション属性は以下のいずれかのみを付与可能です。

jakarta.transaction.Transactional.TxType.REQUIRES_NEW
jakarta.transaction.Transactional.TxType.NOT_SUPPORTED

並列ストリーム操作

ManagedThreadFactory インターフェースが ForkJoinWorkerThreadFactory を継承するようになりました。

public interface ManagedThreadFactory
    extends ThreadFactory, ForkJoinWorkerThreadFactory {
}

コンテキストが伝搬した並列ストリーム操作が可能となります。

ForkJoinWorkerThreadFactory を継承したため、以下のように ForkJoinPool 生成の引数に指定できます。

ManagedThreadFactory managedThreadFactory = //...

ForkJoinPool pool = new ForkJoinPool(
     Runtime.getRuntime().availableProcessors(),
     managedThreadFactory, null, false);

ForkJoinTask<Double> totals = pool.submit(() -> 
     list.parallelStream() ...

Trigger によるスケジュール実行の強化

Trigger を継承した ZonedTriggerインターフェースと、CronTrigger が新しく追加され、Cron 形式でタスク実行スケジュールの定義が可能となりました。

public interface ZonedTrigger extends Trigger {
}

public class CronTrigger implements ZonedTrigger {
}

CronTrigger は以下のように定義できます。

trigger = new CronTrigger("0 7 * SEP-MAY MON-FRI", ZoneId.of("America/New_York"));

fluent に以下のようにすることもできます。

trigger = new CronTrigger(ZoneId.of("America/Los_Angeles"))
           .months(Month.DECEMBER)
           .daysOfMonth(24)
           .hours(16, 18);

実行は今まで通り、ManagedScheduledExecutorService#schedule() にトリガを渡して行います。

その他の変更

jakarta.enterprise.concurrent.spi パッケージが追加され以下のSPIが新しく追加された
- ThreadContextProvider
- ThreadContextRestorer
- ThreadContextSnapshot
- MicroProfile Context Propagation の定義がそのまま移植された形です
CompletionStage、CompletableFuture の Context Propagation
- MicroProfile Context Propagation 仕様からの移植
- org.eclipse.microprofile.context.ThreadContext のメソッドを ContextService にコピー
 - <R> Callable<R> contextualCallable(Callable<R> callable)
 - <T, U> BiConsumer<T, U> contextualConsumer(BiConsumer<T, U> consumer)
 - <T> Consumer<T> contextualConsumer(Consumer<T> consumer)
 - <T, U, R> BiFunction<T, U, R> contextualFunction(BiFunction<T, U, R> function)
 - <T, R> Function<T, R> contextualFunction(Function<T, R> function)
 - Runnable contextualRunnable(Runnable runnable)
 - <R> Supplier<R> contextualSupplier(Supplier<R> supplier)
 - Executor currentContextExecutor()
 - <T> CompletableFuture<T> withContextCapture(CompletableFuture<T> stage)
 - <T> CompletionStage<T> withContextCapture(CompletionStage<T> stage)
- ManagedExecutorService に以下のメソッドが追加された
 -  CompletableFuture completedFuture(U value)
 -  CompletionStage completedStage(U value)
 - <T> CompletableFuture<T> copy(CompletableFuture<T> stage)
 - <T> CompletionStage<T> copy(CompletionStage<T> stage)
 -  CompletableFuture failedFuture(Throwable ex)
 -  CompletionStage failedStage(Throwable ex)
 - public ContextService getContextService()
 -  CompletableFuture newIncompleteFuture()
 - CompletableFuture<Void> runAsync(Runnable runnable)
 -  CompletableFuture supplyAsync(Supplier supplier)