Package org.axonframework.disruptor.commandhandling

Class DisruptorCommandBus

java.lang.Object
org.axonframework.disruptor.commandhandling.DisruptorCommandBus
All Implemented Interfaces:
org.axonframework.commandhandling.CommandBus, org.axonframework.messaging.MessageDispatchInterceptorSupport<org.axonframework.commandhandling.CommandMessage<?>>, org.axonframework.messaging.MessageHandlerInterceptorSupport<org.axonframework.commandhandling.CommandMessage<?>>
public class DisruptorCommandBus extends Object implements org.axonframework.commandhandling.CommandBus
Asynchronous CommandBus implementation with very high performance characteristics. It divides the command handling process in two steps, which can be executed in different threads. The CommandBus is backed by a Disruptor, which ensures that two steps are executed sequentially in these threads, while minimizing locking and inter-thread communication.

The process is split into two separate steps, each of which is executed in a different thread:

  1. Command Handler execution
    This process invokes the command handler with the incoming command. The result and changes to the aggregate are recorded for the next step.
  2. Event storage and publication
    This process stores all generated domain events and publishes them (with any optional application events) to the event bus. Finally, an asynchronous task is scheduled to invoke the command handler callback with the result of the command handling result.
Exceptions and recovery

This separation of process steps makes this implementation very efficient and highly performing. However, it does not cope with exceptions very well. When an exception occurs, an Aggregate that has been loaded is potentially corrupt. That means that an aggregate does not represent a state that can be reproduced by replaying its committed events. Although this implementation will recover from this corrupt state, it may result in a number of commands being rejected in the meantime. These command may be retried if the cause of the AggregateStateCorruptedException does not indicate a non-transient error.

Commands that have been executed against a potentially corrupt Aggregate will result in a AggregateStateCorruptedException exception. These commands are automatically rescheduled for processing by default. Use DisruptorCommandBus.Builder.rescheduleCommandsOnCorruptState(boolean) to disable this feature. Note that the order in which commands are executed is not fully guaranteed when this feature is enabled (default). Limitations of this implementation

Although this implementation allows applications to achieve extreme performance (over 1M commands on commodity hardware), it does have some limitations. It only allows a single aggregate to be invoked during command processing.

This implementation can only work with Event Sourced Aggregates. Infrastructure considerations

This CommandBus implementation has special requirements for the Repositories being used during Command Processing. Therefore, the Repository instance to use in the Command Handler must be created using createRepository(EventStore, AggregateFactory, RepositoryProvider). Using another repository will most likely result in undefined behavior.

The DisruptorCommandBus must have access to at least 3 threads, two of which are permanently used while the DisruptorCommandBus is operational. At least one additional thread is required to invoke callbacks and initiate a recovery process in the case of exceptions.

Consider providing an alternative IdentifierFactory implementation. The default implementation used UUID.randomUUID() to generated identifier for Events. The poor performance of this method severely impacts overall performance of the DisruptorCommandBus. A better performing alternative is, for example, com.eaio.uuid.UUID

Since:
2.0
Author:
Allard Buijze

  • Nested Class Summary

    Nested Classes
    Modifier and Type
    Class
    Description
    static class 
    DisruptorCommandBus.Builder
    Builder class to instantiate a DisruptorCommandBus.

  • Constructor Summary

    Constructors
    Modifier
    Constructor
    Description
    protected
    DisruptorCommandBus(DisruptorCommandBus.Builder builder)
    Instantiate a DisruptorCommandBus based on the fields contained in the DisruptorCommandBus.Builder.

  • Method Summary

    Modifier and Type
    Method
    Description
    static DisruptorCommandBus.Builder
    builder()
    Instantiate a Builder to be able to create a DisruptorCommandBus.
    <T> org.axonframework.modelling.command.Repository<T>
    createRepository(org.axonframework.eventsourcing.eventstore.EventStore eventStore, org.axonframework.eventsourcing.AggregateFactory<T> aggregateFactory)
    Creates a repository instance for an Event Sourced aggregate that is created by the given eventStore and aggregateFactory.
    <T> org.axonframework.modelling.command.Repository<T>
    createRepository(org.axonframework.eventsourcing.eventstore.EventStore eventStore, org.axonframework.eventsourcing.AggregateFactory<T> aggregateFactory, org.axonframework.eventsourcing.SnapshotTriggerDefinition snapshotTriggerDefinition)
    Creates a repository instance for an Event Sourced aggregate, source from given eventStore, that is created by the given aggregateFactory.
    <T> org.axonframework.modelling.command.Repository<T>
    createRepository(org.axonframework.eventsourcing.eventstore.EventStore eventStore, org.axonframework.eventsourcing.AggregateFactory<T> aggregateFactory, org.axonframework.eventsourcing.SnapshotTriggerDefinition snapshotTriggerDefinition, org.axonframework.messaging.annotation.ParameterResolverFactory parameterResolverFactory)
    Creates a repository instance for an Event Sourced aggregate, sourced from given eventStore, that is created by the given aggregateFactory.
    <T> org.axonframework.modelling.command.Repository<T>
    createRepository(org.axonframework.eventsourcing.eventstore.EventStore eventStore, org.axonframework.eventsourcing.AggregateFactory<T> aggregateFactory, org.axonframework.eventsourcing.SnapshotTriggerDefinition snapshotTriggerDefinition, org.axonframework.messaging.annotation.ParameterResolverFactory parameterResolverFactory, org.axonframework.messaging.annotation.HandlerDefinition handlerDefinition, org.axonframework.modelling.command.RepositoryProvider repositoryProvider)
    Creates a repository instance for an Event Sourced aggregate, sourced from given eventStore, that is created by the given aggregateFactory.
    <T> org.axonframework.modelling.command.Repository<T>
    createRepository(org.axonframework.eventsourcing.eventstore.EventStore eventStore, org.axonframework.eventsourcing.AggregateFactory<T> aggregateFactory, org.axonframework.eventsourcing.SnapshotTriggerDefinition snapshotTriggerDefinition, org.axonframework.modelling.command.RepositoryProvider repositoryProvider)
    Creates a repository instance for an Event Sourced aggregate, source from given eventStore, that is created by the given aggregateFactory.
    <T> org.axonframework.modelling.command.Repository<T>
    createRepository(org.axonframework.eventsourcing.eventstore.EventStore eventStore, org.axonframework.eventsourcing.AggregateFactory<T> aggregateFactory, org.axonframework.messaging.annotation.ParameterResolverFactory parameterResolverFactory)
    Creates a repository instance for an Event Sourced aggregate that is created by the given aggregateFactory and sourced from given eventStore.
    <T> org.axonframework.modelling.command.Repository<T>
    createRepository(org.axonframework.eventsourcing.eventstore.EventStore eventStore, org.axonframework.eventsourcing.AggregateFactory<T> aggregateFactory, org.axonframework.messaging.annotation.ParameterResolverFactory parameterResolverFactory, org.axonframework.messaging.annotation.HandlerDefinition handlerDefinition, org.axonframework.modelling.command.RepositoryProvider repositoryProvider)
    Creates a repository instance for an Event Sourced aggregate that is created by the given aggregateFactory and sourced from given eventStore.
    <T> org.axonframework.modelling.command.Repository<T>
    createRepository(org.axonframework.eventsourcing.eventstore.EventStore eventStore, org.axonframework.eventsourcing.AggregateFactory<T> aggregateFactory, org.axonframework.modelling.command.RepositoryProvider repositoryProvider)
    Creates a repository instance for an Event Sourced aggregate that is created by the given eventStore and aggregateFactory.
    <C> void
    dispatch(org.axonframework.commandhandling.CommandMessage<C> command)
     
    <C, R> void
    dispatch(org.axonframework.commandhandling.CommandMessage<C> command, org.axonframework.commandhandling.CommandCallback<? super C,? super R> callback)
     
    org.axonframework.common.Registration
    registerDispatchInterceptor(org.axonframework.messaging.MessageDispatchInterceptor<? super org.axonframework.commandhandling.CommandMessage<?>> dispatchInterceptor)
     
    org.axonframework.common.Registration
    registerHandlerInterceptor(org.axonframework.messaging.MessageHandlerInterceptor<? super org.axonframework.commandhandling.CommandMessage<?>> handlerInterceptor)
     
    void
    stop()
    Shuts down the command bus.
    org.axonframework.common.Registration
    subscribe(String commandName, org.axonframework.messaging.MessageHandler<? super org.axonframework.commandhandling.CommandMessage<?>> handler)
     

    Methods inherited from class java.lang.Object

    clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

  • Constructor Details

    • DisruptorCommandBus

      protected DisruptorCommandBus(DisruptorCommandBus.Builder builder)
      Instantiate a DisruptorCommandBus based on the fields contained in the DisruptorCommandBus.Builder. The Threads required for command execution are immediately requested from the Configuration's Executor, if any. Otherwise, they are created.

      Will assert that the CommandTargetResolver, MessageMonitor, RollbackConfiguration, ProducerType, WaitStrategy and Cache are not null. Additional verification is done on the the coolingDownPeriod, publisherThreadCount, bufferSize and invokerThreadCount to check whether they are positive numbers. If any of these checks fails, an AxonConfigurationException will be thrown.

      Parameters:
      builder - the DisruptorCommandBus.Builder used to instantiate a DisruptorCommandBus instance

  • Method Details

    • builder

      public static DisruptorCommandBus.Builder builder()
      Instantiate a Builder to be able to create a DisruptorCommandBus.

      The following configurable fields have defaults:

      • The rescheduleCommandsOnCorruptState defaults to true.
      • The coolingDownPeriod defaults to 1000.
      • The CommandTargetResolver defaults to an AnnotationCommandTargetResolver.
      • The publisherThreadCount defaults to 1.
      • The MessageMonitor defaults to NoOpMessageMonitor.INSTANCE.
      • The RollbackConfiguration defaults to RollbackConfigurationType.UNCHECKED_EXCEPTIONS.
      • The bufferSize defaults to 4096.
      • The ProducerType defaults to ProducerType.MULTI.
      • The WaitStrategy defaults to a BlockingWaitStrategy.
      • The invokerThreadCount defaults to 1.
      • The Cache defaults to NoCache.INSTANCE.
      • The DuplicateCommandHandlerResolver defaults to DuplicateCommandHandlerResolution.logAndOverride().
      The (2) Threads required for command execution are created immediately. Additional threads are used to invoke response callbacks and to initialize a recovery process in the case of errors. The thread creation process can be specified by providing an Executor.

      The CommandTargetResolver, MessageMonitor, RollbackConfiguration, ProducerType, WaitStrategy and Cache are a hard requirements. Thus setting them to null will result in an AxonConfigurationException. Additionally, the coolingDownPeriod, publisherThreadCount, bufferSize and invokerThreadCount have a positive number constraint, thus will also result in an AxonConfigurationException if set otherwise.

      Returns:
      a Builder to be able to create a DisruptorCommandBus

    • dispatch

      public <C> void dispatch(@Nonnull org.axonframework.commandhandling.CommandMessage<C> command)
      Specified by:
      dispatch in interface org.axonframework.commandhandling.CommandBus

    • dispatch

      public <C, R> void dispatch(@Nonnull org.axonframework.commandhandling.CommandMessage<C> command, @Nonnull org.axonframework.commandhandling.CommandCallback<? super C,? super R> callback)
      Specified by:
      dispatch in interface org.axonframework.commandhandling.CommandBus

    • createRepository

      public <T> org.axonframework.modelling.command.Repository<T> createRepository(org.axonframework.eventsourcing.eventstore.EventStore eventStore, org.axonframework.eventsourcing.AggregateFactory<T> aggregateFactory)
      Creates a repository instance for an Event Sourced aggregate that is created by the given eventStore and aggregateFactory.

      The repository returned must be used by Command Handlers subscribed to this Command Bus for loading aggregate instances. Using any other repository instance may result in undefined outcome (a.k.a. concurrency problems).

      Type Parameters:
      T - The type of aggregate to create the repository for
      Parameters:
      eventStore - The Event Store to retrieve and persist events
      aggregateFactory - The factory creating uninitialized instances of the Aggregate
      Returns:
      the repository that provides access to stored aggregates

    • createRepository

      public <T> org.axonframework.modelling.command.Repository<T> createRepository(org.axonframework.eventsourcing.eventstore.EventStore eventStore, org.axonframework.eventsourcing.AggregateFactory<T> aggregateFactory, org.axonframework.modelling.command.RepositoryProvider repositoryProvider)
      Creates a repository instance for an Event Sourced aggregate that is created by the given eventStore and aggregateFactory.

      The repository returned must be used by Command Handlers subscribed to this Command Bus for loading aggregate instances. Using any other repository instance may result in undefined outcome (a.k.a. concurrency problems).

      Type Parameters:
      T - The type of aggregate to create the repository for
      Parameters:
      eventStore - The Event Store to retrieve and persist events
      aggregateFactory - The factory creating uninitialized instances of the Aggregate
      repositoryProvider - Provides repositories for specified aggregate types
      Returns:
      the repository that provides access to stored aggregates

    • createRepository

      public <T> org.axonframework.modelling.command.Repository<T> createRepository(org.axonframework.eventsourcing.eventstore.EventStore eventStore, org.axonframework.eventsourcing.AggregateFactory<T> aggregateFactory, org.axonframework.eventsourcing.SnapshotTriggerDefinition snapshotTriggerDefinition)
      Creates a repository instance for an Event Sourced aggregate, source from given eventStore, that is created by the given aggregateFactory.

      The repository returned must be used by Command Handlers subscribed to this Command Bus for loading aggregate instances. Using any other repository instance may result in undefined outcome (a.k.a. concurrency problems).

      Type Parameters:
      T - The type of aggregate to create the repository for
      Parameters:
      eventStore - The Event Store to retrieve and persist events
      aggregateFactory - The factory creating uninitialized instances of the Aggregate
      snapshotTriggerDefinition - The trigger definition for creating snapshots
      Returns:
      the repository that provides access to stored aggregates

    • createRepository

      public <T> org.axonframework.modelling.command.Repository<T> createRepository(org.axonframework.eventsourcing.eventstore.EventStore eventStore, org.axonframework.eventsourcing.AggregateFactory<T> aggregateFactory, org.axonframework.eventsourcing.SnapshotTriggerDefinition snapshotTriggerDefinition, org.axonframework.modelling.command.RepositoryProvider repositoryProvider)
      Creates a repository instance for an Event Sourced aggregate, source from given eventStore, that is created by the given aggregateFactory. The given decorator is used to decorate event streams.

      The repository returned must be used by Command Handlers subscribed to this Command Bus for loading aggregate instances. Using any other repository instance may result in undefined outcome (a.k.a. concurrency problems).

      Note that a second invocation of this method with an aggregate factory for the same aggregate type may return the same instance as the first invocation, even if the given decorator is different.

      Type Parameters:
      T - The type of aggregate to create the repository for
      Parameters:
      eventStore - The Event Store to retrieve and persist events
      aggregateFactory - The factory creating uninitialized instances of the Aggregate
      snapshotTriggerDefinition - The trigger definition for creating snapshots
      repositoryProvider - Provides repositories for specified aggregate types
      Returns:
      the repository that provides access to stored aggregates

    • createRepository

      public <T> org.axonframework.modelling.command.Repository<T> createRepository(org.axonframework.eventsourcing.eventstore.EventStore eventStore, org.axonframework.eventsourcing.AggregateFactory<T> aggregateFactory, org.axonframework.messaging.annotation.ParameterResolverFactory parameterResolverFactory)
      Creates a repository instance for an Event Sourced aggregate that is created by the given aggregateFactory and sourced from given eventStore. Parameters of the annotated methods are resolved using the given parameterResolverFactory.
      Type Parameters:
      T - The type of aggregate managed by this repository
      Parameters:
      eventStore - The Event Store to retrieve and persist events
      aggregateFactory - The factory creating uninitialized instances of the Aggregate
      parameterResolverFactory - The ParameterResolverFactory to resolve parameter values of annotated handler with
      Returns:
      the repository that provides access to stored aggregates

    • createRepository

      public <T> org.axonframework.modelling.command.Repository<T> createRepository(org.axonframework.eventsourcing.eventstore.EventStore eventStore, org.axonframework.eventsourcing.AggregateFactory<T> aggregateFactory, org.axonframework.messaging.annotation.ParameterResolverFactory parameterResolverFactory, org.axonframework.messaging.annotation.HandlerDefinition handlerDefinition, org.axonframework.modelling.command.RepositoryProvider repositoryProvider)
      Creates a repository instance for an Event Sourced aggregate that is created by the given aggregateFactory and sourced from given eventStore. Parameters of the annotated methods are resolved using the given parameterResolverFactory. The given handlerDefinition is used to create handler instances.
      Type Parameters:
      T - The type of aggregate managed by this repository
      Parameters:
      eventStore - The Event Store to retrieve and persist events
      aggregateFactory - The factory creating uninitialized instances of the Aggregate
      parameterResolverFactory - The ParameterResolverFactory to resolve parameter values of annotated handler with
      handlerDefinition - The handler definition used to create concrete handlers
      repositoryProvider - Provides specific for given aggregate types
      Returns:
      the repository that provides access to stored aggregates

    • createRepository

      public <T> org.axonframework.modelling.command.Repository<T> createRepository(org.axonframework.eventsourcing.eventstore.EventStore eventStore, org.axonframework.eventsourcing.AggregateFactory<T> aggregateFactory, org.axonframework.eventsourcing.SnapshotTriggerDefinition snapshotTriggerDefinition, org.axonframework.messaging.annotation.ParameterResolverFactory parameterResolverFactory)
      Creates a repository instance for an Event Sourced aggregate, sourced from given eventStore, that is created by the given aggregateFactory. Parameters of the annotated methods are resolved using the given parameterResolverFactory.
      Type Parameters:
      T - The type of aggregate managed by this repository
      Parameters:
      eventStore - The Event Store to retrieve and persist events
      aggregateFactory - The factory creating uninitialized instances of the Aggregate
      snapshotTriggerDefinition - The trigger definition for snapshots
      parameterResolverFactory - The ParameterResolverFactory to resolve parameter values of annotated handler with
      Returns:
      the repository that provides access to stored aggregates

    • createRepository

      public <T> org.axonframework.modelling.command.Repository<T> createRepository(org.axonframework.eventsourcing.eventstore.EventStore eventStore, org.axonframework.eventsourcing.AggregateFactory<T> aggregateFactory, org.axonframework.eventsourcing.SnapshotTriggerDefinition snapshotTriggerDefinition, org.axonframework.messaging.annotation.ParameterResolverFactory parameterResolverFactory, org.axonframework.messaging.annotation.HandlerDefinition handlerDefinition, org.axonframework.modelling.command.RepositoryProvider repositoryProvider)
      Creates a repository instance for an Event Sourced aggregate, sourced from given eventStore, that is created by the given aggregateFactory. Parameters of the annotated methods are resolved using the given parameterResolverFactory. The given handlerDefinition is used to create handler instances.
      Type Parameters:
      T - The type of aggregate managed by this repository
      Parameters:
      eventStore - The Event Store to retrieve and persist events
      aggregateFactory - The factory creating uninitialized instances of the Aggregate
      snapshotTriggerDefinition - The trigger definition for snapshots
      parameterResolverFactory - The ParameterResolverFactory to resolve parameter values of annotated handler with
      handlerDefinition - The handler definition used to create concrete handlers
      repositoryProvider - Provides repositories for specific aggregate types
      Returns:
      the repository that provides access to stored aggregates

    • subscribe

      public org.axonframework.common.Registration subscribe(@Nonnull String commandName, @Nonnull org.axonframework.messaging.MessageHandler<? super org.axonframework.commandhandling.CommandMessage<?>> handler)
      Specified by:
      subscribe in interface org.axonframework.commandhandling.CommandBus

    • stop

      public void stop()
      Shuts down the command bus. It no longer accepts new commands, and finishes processing commands that have already been published. This method will not shut down any executor that has been provided as part of the Builder process.

    • registerDispatchInterceptor

      @Nonnull public org.axonframework.common.Registration registerDispatchInterceptor(@Nonnull org.axonframework.messaging.MessageDispatchInterceptor<? super org.axonframework.commandhandling.CommandMessage<?>> dispatchInterceptor)
      Specified by:
      registerDispatchInterceptor in interface org.axonframework.messaging.MessageDispatchInterceptorSupport<org.axonframework.commandhandling.CommandMessage<?>>

    • registerHandlerInterceptor

      public org.axonframework.common.Registration registerHandlerInterceptor(@Nonnull org.axonframework.messaging.MessageHandlerInterceptor<? super org.axonframework.commandhandling.CommandMessage<?>> handlerInterceptor)
      Specified by:
      registerHandlerInterceptor in interface org.axonframework.messaging.MessageHandlerInterceptorSupport<org.axonframework.commandhandling.CommandMessage<?>>