メインリクエストのレスポンス時間を短縮しつつ、監査ログの出力や通知送信などの後続処理をバックグラウンドで実行したいケースは多い。Guavaライブラリが提供するAsyncEventBusを利用すれば、Pub/Subモデルに基づいた非同期イベント処理を簡潔に実装できる。コアロジックと副次処理を分離し、拡張性を保つためのSpring Boot統合手順を以下に示す。

1. 依存関係の追加

MavenプロジェクトにGuavaの依存関係を追加する。

<dependency>
    <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
    <version>32.1.3-jre</version>
</dependency>

2. 非同期実行用スレッドプールの定義

イベントハンドラーが動作するスレッドプールを構成する。ここではThreadPoolTaskExecutorを使用し、キューの飽和時に発生するリソース枯渇を防ぐためCallerRunsPolicyを適用する。

@Configuration
public class AsyncExecutorConfiguration {

    @Bean(name = "eventBusExecutor")
    public ThreadPoolTaskExecutor eventExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setThreadNamePrefix("guava-event-");
        executor.setCorePoolSize(Runtime.getRuntime().availableProcessors());
        executor.setMaxPoolSize(Runtime.getRuntime().availableProcessors() * 2);
        executor.setQueueCapacity(200);
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }
}

3. イベント処理例外ハンドラーの実装

非同期スレッド内で例外が発生した場合、呼び出し元スレッドには伝播しないため、専用のハンドラーでエラーを捕捉・記録する必要がある。

@Service
public class EventBusErrorHandler implements SubscriberExceptionHandler {
    private static final Logger log = LoggerFactory.getLogger(EventBusErrorHandler.class);

    @Override
    public void handleException(Throwable thrownException, SubscriberExceptionContext context) {
        log.error("非同期イベント処理中に予期せぬエラーが発生しました。イベントタイプ: {}, 原因: {}",
                  context.getMethod().getName(), thrownException.getMessage(), thrownException);
    }
}

4. AsyncEventBusインスタンスの生成

前述のExecutorと例外ハンドラーを紐付け、イベントバス本体をSpring Beanとして登録する。

@Configuration
public class EventBusCoreConfig {

    private final ThreadPoolTaskExecutor eventExecutor;
    private final EventBusErrorHandler errorHandler;

    public EventBusCoreConfig(
            @Qualifier("eventBusExecutor") ThreadPoolTaskExecutor eventExecutor,
            EventBusErrorHandler errorHandler) {
        this.eventExecutor = eventExecutor;
        this.errorHandler = errorHandler;
    }

    @Bean
    public AsyncEventBus applicationEventBus() {
        return new AsyncEventBus(eventExecutor, errorHandler);
    }
}

5. イベントオブジェクトとサブスクライバーの設計

まず、任意のデータをカプセル化するジェネリクス型イベントクラスを定義する。

public class DomainEvent<T> {
    private final T payload;
    private final Instant occurredAt;

    public DomainEvent(T payload) {
        this.payload = payload;
        this.occurredAt = Instant.now();
    }

    public T getPayload() { return payload; }
    public Instant getOccurredAt() { return occurredAt; }
}

次に、イベントを受信して処理を行うリスナークラスを作成する。@Subscribeアノテーションが付与されたメソッドが、イベント発行時に自動的に呼び出される。

@Component
public class NotificationSubscriber {
    private static final Logger log = LoggerFactory.getLogger(NotificationSubscriber.class);

    @Subscribe
    @AllowConcurrentEvents
    public void handleSystemEvent(DomainEvent<String> event) {
        // 非同期バックグラウンドで実行される処理
        log.info("非同期イベントを受信しました。ペイロード: {}, 発生時刻: {}",
                 event.getPayload(), event.getOccurredAt());
        // ここでDBへのログ書き込みや外部API呼び出しなどを実装
    }
}

6. サブスクライバーのイベントバスへの登録

Springコンテキストの初期化完了後に、生成されたリスナーをAsyncEventBusへ登録する。InitializingBeanインタフェースを実装することで、Beanの依存関係解決直後に安全に登録処理を実行できる。

@Component
public class SubscriberRegistry implements InitializingBean {
    private final AsyncEventBus eventBus;
    private final NotificationSubscriber subscriber;

    public SubscriberRegistry(AsyncEventBus eventBus, NotificationSubscriber subscriber) {
        this.eventBus = eventBus;
        this.subscriber = subscriber;
    }

    @Override
    public void afterPropertiesSet() {
        eventBus.register(subscriber);
    }
}

7. イベント発行用エンドポイントの検証

実際にイベントを発行し、非同期コンシューマーが動作するかを確認するためのテストコントローラーを配置する。

@RestController
@RequestMapping("/api/v1/events")
public class EventPublishController {

    private final AsyncEventBus asyncEventBus;

    public EventPublishController(AsyncEventBus asyncEventBus) {
        this.asyncEventBus = asyncEventBus;
    }

    @PostMapping("/trigger")
    public ResponseEntity<String> triggerEvent(@RequestParam String content) {
        DomainEvent<String> payloadEvent = new DomainEvent<>(content);
        // メインスレッドは即座にレスポンスを返す
        asyncEventBus.post(payloadEvent);
        return ResponseEntity.ok("EVENT_DISPATCHED");
    }
}

この構成により、/api/v1/events/triggerへのリクエスト処理が完了した直後、定義されたスレッドプール上でバックグラウンド処理が実行される。イベントバスに対して追加のDomainEvent型や@Subscribeメソッドを用意すれば、複数のサブスクライバーを並列に動作させることが可能となる。