Use domain events to capture something that happens in a domain.
Domain-driven practitioners find that they can better understand the problem domain by learning more about the events that occur in the problem domain. These events, namely domain events, are mainly obtained in the process of knowledge extraction with domain experts.
Domain events, which can be used within a domain model within a bounded context, or message queues can be used for asynchronous communication between bounded contexts.
1 Understand domain events
Domain events are events that occur in a domain that are of interest to domain experts.
Activities occurring in a field are modeled as a series of discrete events. Each event is represented by a domain object. Domain events are part of the domain model and represent what happens in the domain.
Main uses of domain events:
- Ensure data consistency across aggregations
- Replacement batch processing
- Implement the event source pattern
- Perform bounded context integration
Implement domain events
A domain event is a fact that has already happened and will not change after it has happened. Therefore, domain events are typically modeled as value objects.
However, there are special cases where modeling compromises are often made to meet the needs of serialization and deserialization frameworks.
2.1 Creating domain events
2.1.1 Event named
When modeling domain events, we should name the events according to the common language in the bounded context.
If an event is generated by a command action on an aggregation, the event is usually named after the name of that action method. The event name indicates the fact that the command method on the aggregation was successfully executed. That is, event names need to reflect what happened in the past.
public class AccountEnabledEvent extends AbstractAggregateEvent<Long, Account> {
public AccountEnabledEvent(Account source) {
super(source); }}Copy the code2.1.2 Event attributes
The attributes of the event are primarily used to drive subsequent business processes. Of course, it also has some common properties.
Events have some common properties, such as:
- A unique identifier
- OccurredOn occurred
- Type Event type
- Source Source of event occurrence (only for events generated by aggregation)
Generic properties can be regulated using the event interface.
| Interface or class | meaning |
|---|---|
| DomainEvent | Generic domain event interface |
| AggregateEvent | Common domain event interface published by aggregation |
| AbstractDomainEvent | DomainEvent implementation class, maintenance ID and creation time |
| AbstractAggregateEvent | The AggregateEvent implementation class inherits AbstractDomainEvent and adds the source property |
However, events are primarily business attributes. We need to consider who caused the event to happen, which could involve the aggregation that produced the event or other aggregation that participated in the operation, or any other type of operational data.
2.1.3 Event method
Events are factual descriptions that do not have much business action on their own.
Domain events are typically designed as immutable objects, and the data they carry already reflects the source of the event. The event constructor performs state initialization and provides getter methods for the property.
2.1.4 Event unique identifier
The important thing to note here is the unique identifier of events. In general, events are immutable, so why the concept of unique identifier?
For domain events published from aggregation, the name of the event, the identifier that generated the event, the time when the event occurred, and so on are sufficient to distinguish between different events. However, this can increase the complexity of comparing events.
For events published by the caller, we model the domain events as aggregates, using the unique identity of the aggregate directly as the identification of the event.
The introduction of event unique identifier will greatly reduce the complexity of event comparison. However, its greatest significance lies in the integration of bounded contexts.
When we need to publish domain events to external bounded contexts, unique identifiers are a necessity. In order to ensure the idempotency of event delivery, at the sending end, we may make several attempts to send the event until it is clearly sent successfully. On the receiving end, repeatability detection is needed to ensure idempotency of event processing after receiving the event. In this case, the unique identifier of the event can be used as the basis for event de-duplication.
Event unique identification, by itself, has little impact on domain modeling, but is of great benefit to technical processing. Therefore, it is managed as a generic property.
2.2 Publishing domain events
How do we avoid coupling domain events with handlers?
A simple and efficient way to do this is to use the observer pattern, which decouples domain events from external components.
2.2.1 Publish and subscribe model
To be consistent, we need to define a set of interfaces and implementation classes for publishing events based on the observer pattern.
The interfaces and implementation classes involved are as follows:
| Interface or class | meaning |
|---|---|
| DomainEventPublisher | Used to publish domain events |
| DomainEventHandlerRegistry | Use to register DomainEventHandler |
| DomainEventBus | Extend the DomainEventPublisher event handlers and DomainEventHandlerRegistry used to publish and management field |
| DefaultDomainEventBus | DomainEventBus implemented by default |
| DomainEventHandler | For handling domain events |
| DomainEventSubscriber | Used to determine whether to accept domain events |
| DomainEventExecutor | Used to execute the domain event handler |
Using the real example shown in DomainEventBusTest:
public class DomainEventBusTest {
private DomainEventBus domainEventBus;
@Before
public void setUp() throws Exception {
this.domainEventBus = new DefaultDomainEventBus();
}
@After
public void tearDown() throws Exception {
this.domainEventBus = null;
}
@Test
public void publishTest(){// create eventHandler TestEventHandler eventHandler = new TestEventHandler(); / / register event handlers enclosing domainEventBus. Register (TestEvent. Class, eventHandler); / / publish event this. DomainEventBus. The publish (new TestEvent ("123")); // Check that the event handler is enough to run assert.assertequals ("123", eventHandler.data); } @Value class TestEvent extends AbstractDomainEvent{ private String data; } class TestEventHandler implements DomainEventHandler<TestEvent>{ private String data; @Override public void handle(TestEvent event) { this.data = event.getData(); }}}Copy the codeOnce the publish-subscribe structure is built, it needs to be associated with the domain model. How the domain model gets Publisher, and how the event handler subscribes.
2.2.2 Event publishing based on ThreadLocal
A common solution is to bind DomainEventBus to the thread context. This way, as long as the same calling thread can easily get the DomainEventBus object.
Specific interactions are as follows:
DomainEventBusHolder Manages DomainEventBus.
public class DomainEventBusHolder {
private static final ThreadLocal<DomainEventBus> THREAD_LOCAL = new ThreadLocal<DomainEventBus>(){
@Override
protected DomainEventBus initialValue() {
returnnew DefaultDomainEventBus(); }}; public static DomainEventPublishergetPubliser() {return THREAD_LOCAL.get();
}
public static DomainEventHandlerRegistry getHandlerRegistry() {return THREAD_LOCAL.get();
}
public static void clean(){ THREAD_LOCAL.remove(); }}Copy the codeThe enable Account is directly published using DomainEventBusHolder.
public class Account extends JpaAggregate {
public void enable(){
AccountEnabledEvent event = new AccountEnabledEvent(this);
DomainEventBusHolder.getPubliser().publish(event);
}
}
public class AccountEnabledEvent extends AbstractAggregateEvent<Long, Account> {
public AccountEnabledEvent(Account source) {
super(source); }}Copy the codeThe AccountApplication does the subscriber registration and business method invocation.
public class AccountApplication extends AbstractApplication {
private static final Logger LOGGER = LoggerFactory.getLogger(AccountApplication.class);
@Autowired
private AccountRepository repository;
public void enable(id) {/ / clean up before binding Handler DomainEventBusHolder. Clean (); / / registered EventHandler AccountEnableEventHandlerenableEventHandler = new AccountEnableEventHandler();
DomainEventBusHolder.getHandlerRegistry().register(AccountEnabledEvent.class, enableEventHandler);
Optional<Account> accountOptional = repository.getById(id);
if (accountOptional.isPresent()) {
Account account = accountOptional.get();
// enablePublish the event account.enable() directly using DomainEventBusHolder; repository.save(account); } } class AccountEnableEventHandler implements DomainEventHandler<AccountEnabledEvent>{ @Override public void handle(AccountEnabledEvent event) { LOGGER.info("handle enable event"); }}}Copy the code2.2.3 Event publishing based on entity cache
Events are first cached in entities and then published after the entity state is successfully persisted to storage.
Specific interactions are as follows:
Example code is as follows:
public class Account extends JpaAggregate {
public void enable(){ AccountEnabledEvent event = new AccountEnabledEvent(this); registerEvent(event); }}Copy the codeThe Enable method of Account calls registerEvent to register events.
@MappedSuperclass
public abstract class AbstractAggregate<ID> extends AbstractEntity<ID> implements Aggregate<ID> {
private static final Logger LOGGER = LoggerFactory.getLogger(AbstractAggregate.class);
@JsonIgnore
@QueryTransient
@Transient
@org.springframework.data.annotation.Transient
private final transient List<DomainEventItem> events = Lists.newArrayList();
protected void registerEvent(DomainEvent event) {
events.add(new DomainEventItem(event));
}
protected void registerEvent(Supplier<DomainEvent> eventSupplier) {
this.events.add(new DomainEventItem(eventSupplier));
}
@Override
@JsonIgnore
public List<DomainEvent> getEvents() {
return Collections.unmodifiableList(events.stream()
.map(eventSupplier -> eventSupplier.getEvent())
.collect(Collectors.toList()));
}
@Override
public void cleanEvents() { events.clear(); } private class DomainEventItem { DomainEventItem(DomainEvent event) { Preconditions.checkArgument(event ! = null); this.domainEvent = event; } DomainEventItem(Supplier<DomainEvent> supplier) { Preconditions.checkArgument(supplier ! = null); this.domainEventSupplier = supplier; } private DomainEvent domainEvent; private Supplier<DomainEvent> domainEventSupplier; public DomainEventgetEvent() {
if(domainEvent ! = null) {returndomainEvent; } DomainEvent event = this.domainEventSupplier ! = null ? this.domainEventSupplier.get() : null; domainEvent = event;returndomainEvent; }}}Copy the codeIn AbstractAggregate, the registerEvent method stores events to the events collection, the getEvents method retrieves all events, and the cleanEvents method cleans up cached events.
The Application example is as follows:
@Service
public class AccountApplication extends AbstractApplication {
private static final Logger LOGGER = LoggerFactory.getLogger(AccountApplication.class);
@Autowired
private AccountRepository repository;
@Autowired
private DomainEventBus domainEventBus;
@PostConstruct
public void init() {/ / use the Spring life cycle register event handlers enclosing domainEventBus. Register (AccountEnabledEvent. Class, new AccountEnableEventHandler ()); } public voidenable(Long id){
Optional<Account> accountOptional = repository.getById(id);
if (accountOptional.isPresent()) {
Account account = accountOptional.get();
// enableCache events in account.enable(); repository.save(account); List<DomainEvent> events = account.getEvents();if(! CollectionUtils. IsEmpty (events)) {/ / after successful persistence, to release this. Events domainEventBus. PublishAll (events); } } } class AccountEnableEventHandler implements DomainEventHandler<AccountEnabledEvent>{ @Override public void handle(AccountEnabledEvent event) { LOGGER.info("handle enable event"); }}}Copy the codeThe Init method of AccountApplication registers event listeners, and the Enable method publishes cached events through the DomainEventBus instance after the entity is successfully persisted.
2.2.4 Events are published by the caller
Typically, domain events are generated by aggregated command methods and published after successful command method execution. Sometimes domain events are not generated by command methods in the aggregation, but by requests made by the user.
At this point, we need to model domain events as an aggregation and have our own repository. However, because domain events represent events that happened in the past, the repository only does appending and cannot modify or delete events.
For example, publish user click events.
@Entity @Data public class ClickAction extends JpaAggregate implements DomainEvent { @Setter(AccessLevel.PRIVATE) private Long userId; @Setter(AccessLevel.PRIVATE) private String menuId; public ClickAction(Long userId, String menuId){ Preconditions.checkArgument(userId ! = null); Preconditions.checkArgument(StringUtils.isNotEmpty(menuId));setUserId(userId);
setMenuId(menuId);
}
@Override
public String id() {
return String.valueOf(getId());
}
@Override
public Date occurredOn() {
returngetCreateTime(); }}Copy the codeClickAction inherits from the JpaAggregate implementation of the DomainEvent interface and overwrites the ID and occurredOn methods.
@Service public class ClickActionApplication extends AbstractApplication { @Autowired private ClickActionRepository repository; @Autowired private DomainEventBus domainEventBus; public void clickMenu(Long id, String menuId){ ClickAction clickAction = new ClickAction(id, menuId); clickAction.prePersist(); this.repository.save(clickAction); domainEventBus.publish(clickAction); }}Copy the codeClickActionApplication publishes the event using DomainEventBus after successfully saving the ClickAction.
2.3 Subscribe to domain events
What component registers a subscriber with a domain event? Most requests are made by application services and can sometimes be registered by domain services.
Because the application service is a direct customer of the domain model, it is an ideal place to register a domain event subscriber, that is, to subscribe to events before the application service invokes the domain methods.
Subscribe based on ThreadLocal:
public void enable(id) {/ / clean up before binding Handler DomainEventBusHolder. Clean (); / / registered EventHandler AccountEnableEventHandlerenableEventHandler = new AccountEnableEventHandler();
DomainEventBusHolder.getHandlerRegistry().register(AccountEnabledEvent.class, enableEventHandler);
Optional<Account> accountOptional = repository.getById(id);
if (accountOptional.isPresent()) {
Account account = accountOptional.get();
// enablePublish the event account.enable() directly using DomainEventBusHolder; repository.save(account); }}Copy the codeSubscribe based on entity cache:
@PostConstruct
public void init() {/ / use the Spring life cycle register event handlers enclosing domainEventBus. Register (AccountEnabledEvent. Class, new AccountEnableEventHandler ()); } public voidenable(Long id){
Optional<Account> accountOptional = repository.getById(id);
if (accountOptional.isPresent()) {
Account account = accountOptional.get();
// enableCache events in account.enable(); repository.save(account); List<DomainEvent> events = account.getEvents();if(! CollectionUtils. IsEmpty (events)) {/ / after successful persistence, to release this. Events domainEventBus. PublishAll (events); }}}Copy the code2.4 Handling domain events
With the event published, let’s take a look at event handling.
Against 2.4.1Ensure data consistency across aggregations
We typically use domain events to maintain model consistency. One of the principles in aggregation modeling is that only one aggregation can be modified in a transaction, and the resulting changes must run in a separate transaction.
In this case, care needs to be taken about the transmissibility of the transaction.
Application services control transactions. Do not modify another aggregation instance in the middle of event notification, as this breaks one of the aggregation principles: only one aggregation can be modified in a transaction.
For simple scenarios, we can isolate aggregate changes using a special transaction isolation policy. The specific process is as follows:
However, the best solution is to use asynchronous processing. And each definer modifies additional aggregation instances in their own separate transactions.
Event subscribers should not execute command methods on another aggregation because doing so would break the principle of only modifying a single aggregation instance in a single transaction. The final consistency between all aggregation instances must be handled asynchronously.
See asynchronously handling domain events.
2.4.2 Replacement batch processing
Batch processes typically require complex queries and require large transaction support. If the system processes domain events as soon as they are received, business requirements are not only met more quickly, but batch operations are eliminated.
During off-peak periods, batch processing is commonly used for system maintenance, such as deleting outdated data, creating new objects, notifying users, updating statistics, and so on. These batch processes often require complex queries and require large transaction support.
If we listen for a domain event in the system, the system handles it immediately when we receive a domain event. In this way, the original batch centralized processing process is scattered into many small processing units, service needs can be met faster, and users can timely proceed with the next operation.
2.4.3 Implement the event source pattern
There are many benefits to maintaining an event store for all domain events in a single bounded context.
Storing events allows you to:
- The event store is used as a message queue, and domain events are then published through the messaging facility.
- Store events for reST-based event notification.
- Check the history of the results of the model naming method.
- Use event stores for business prediction and analysis.
- Use events to rebuild the aggregation instance.
- Perform the undo operation of the aggregation.
Event storage is a big topic that will be covered in a chapter.
2.4.4 Perform bounded context integration
The bounded context integration based on domain events mainly consists of two modes: message queue and REST event.
Here, the focus is on context integration based on message queues.
When adopting messaging systems in different contexts, we must ensure ultimate consistency. In this case, we need to preserve final consistency between at least two stores: the store used by the domain model and the persistent store used by the message queue. We must ensure that when we persist the domain model, the events for have also been successfully published. If the two are out of sync, the model may be in an incorrect state.
In general, there are three ways:
- Domain model and message sharing persistent store. In this case, the submission of the model and the event is done in a single transaction to ensure consistency between the two.
- Domain models and messages are controlled by global transactions. In this case, the persistent storage used by the model and the message can be separated, but can degrade system performance.
- In a domain persistent store, a special storage area is created to store events (that is, event stores), thus completing the storage of the domain and events in a local transaction. Events are then sent asynchronously to a message queue via a background service.
The third, in general, is the more elegant solution.
When consistency is not high, events can be sent directly to the message queue through the domain event subscriber. The specific process is as follows:
When consistency is high, events need to be stored and then loaded and distributed to message queues via background threads. The specific process is as follows:
2.5 Handle domain events asynchronously
Domain events can work with asynchronous workflows, including asynchronous communication between bounded contexts using message queues. Of course, asynchronous processing can also be started in the same bounded context.
As an event publisher, you should not care if asynchronous processing is performed. Exception handling is determined by the event executor.
DomainEventExecutor provides support for asynchronous processing.
DomainEventExecutor eventExecutor =
new ExecutorBasedDomainEventExecutor("EventHandler", 1, 100);
this.domainEventBus.register(AccountEnabledEvent.class,
eventExecutor,
new AccountEnableEventHandler());
Copy the codeAsynchronous processing means abandoning the ACID nature of database transactions in favor of final consistency.
2.6 Internal Events and External Events
When domain events are used, they need to be distinguished to avoid technical implementation problems.
It is important to recognize the difference between internal and external events.
- Internal events are events within a domain model that are not shared between bounded contexts.
- External events are published events that are shared in multiple bounded contexts.
Typically, in a typical business use case, there may be many internal events and only one or two external events.
2.6.1 Internal events
An inner event exists inside a bounded context and is protected by a bounded context boundary.
Internal events are restricted within a single bounded context boundary, so domain objects can be referenced directly.
public interface AggregateEvent<ID, A extends Aggregate<ID>> extends DomainEvent{
A source(a); default AgetSource() {return source();
}
}
Copy the codeFor example, the source in an AggregateEvent points to the aggregation that publishes the event.
public class LikeSubmittedEvent extends AbstractAggregateEvent<Long, Like> {
public LikeSubmittedEvent(Like source) {
super(source);
}
public LikeSubmittedEvent(String id, Like source) {
super(id, source); }}Copy the codeThe LikeSubmittedEvent class directly references the Like aggregation.
2.6.2 External events
External events exist between bounded contexts and are shared by multiple contexts.
In general, external events exist only as data carriers. A flat structure is often used and all attributes are exposed.
@Data
public class SubmittedEvent {
private Owner owner;
private Target target;
}
Copy the codeSubmittedEvent is a flat structure, which mainly encapsulates data.
Because external events are shared by multiple contexts, versioning is important to avoid significant changes affecting their services.
Implement the domain event pattern
Domain events are a generic pattern that essentially adds domain concepts to the publish-subscribe pattern.
3.1 A publish-subscribe pattern that encapsulates domain events
Publish – subscribe is a mature design pattern with high universality. Therefore, encapsulation is recommended for domain requirements.
For example, use geekhalo-DDD-related modules directly.
Define domain events:
@Value
public class LikeCancelledEvent extends AbstractAggregateEvent<Long, Like> {
public LikeCancelledEvent(Like source) {
super(source); }}Copy the codeSubscribe to domain events:
this.domainEventBus.register(LikeCancelledEvent.class, likeCancelledEvent->{
CanceledEvent canceledEvent = new CanceledEvent();
canceledEvent.setOwner(likeCancelledEvent.getSource().getOwner());
canceledEvent.setTarget(likeCancelledEvent.getSource().getTarget());
this.redisBasedQueue.pushLikeEvent(canceledEvent);
});
Copy the codeAsynchronously execute domain events:
DomainEventExecutor eventExecutor =
new ExecutorBasedDomainEventExecutor("LikeEventHandler", 1, 100);
this.domainEventBus.register(LikeCancelledEvent.class,
eventExecutor,
likeCancelledEvent->{
CanceledEvent canceledEvent = new CanceledEvent();
canceledEvent.setOwner(likeCancelledEvent.getSource().getOwner());
canceledEvent.setTarget(likeCancelledEvent.getSource().getTarget());
this.redisBasedQueue.pushLikeEvent(canceledEvent);
});
Copy the code3.2 The memory bus handles internal events and the message queue handles external events
Memory bus is simple and efficient, and supports synchronous and asynchronous processing schemes, which is suitable for handling complex internal events. Message queues, while complex, are good at solving inter-service communication problems and are suitable for handling external events.
3.3 Use entity caching for domain events
In theory, events should be published only after the business has successfully completed. Therefore, caching domain events in entities and publishing them after completing business operations is a better solution.
The event caching scheme has significant advantages over using ThreadLocal to manage subscribers and subscribe to callbacks when events are published.
3.4 Using the event publishing function of the IOC container
The IOC container provides a lot of functionality to use, including publish-subscribe functionality such as Spring.
In general, domain models should not rely directly on the Spring container. Therefore, we still use the memory bus in the realm, adding a subscriber to it and forwarding events from the memory bus to the Spring container.
class SpringEventDispatcher implements ApplicationEventPublisherAware {
@Autowired
private DomainEventBus domainEventBus;
private ApplicationEventPublisher eventPublisher;
@Override
public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
this.eventPublisher = applicationEventPublisher;
}
@PostConstruct
public void addListener(){
this.domainEventBus.register(event->true, event -> {this.eventPublisher.publishEvent(event);});
}
}
Copy the codeAt this point, we can handle domain events directly using Spring’s EventListener mechanism.
@Component public class RedisBasedQueueExporter { @Autowired private RedisBasedQueue redisBasedQueue; @EventListener public void handle(LikeSubmittedEvent likeSubmittedEvent){ SubmittedEvent submittedEvent = new SubmittedEvent(); submittedEvent.setOwner(likeSubmittedEvent.getSource().getOwner()); submittedEvent.setTarget(likeSubmittedEvent.getSource().getTarget()); this.redisBasedQueue.pushLikeEvent(submittedEvent); } @EventListener public void handle(LikeCancelledEvent likeCancelledEvent){ CanceledEvent canceledEvent = new CanceledEvent(); canceledEvent.setOwner(likeCancelledEvent.getSource().getOwner()); canceledEvent.setTarget(likeCancelledEvent.getSource().getTarget()); this.redisBasedQueue.pushLikeEvent(canceledEvent); }}Copy the code4 summary
- Domain events are facts that occur in the problem domain and are part of the common language.
- Domain events use publish-subscribe preferentially, publishing events and firing the appropriate event handlers.
- In bounded context, internal events and memory bus are preferred; Between bounded contexts, external event and message queues are preferred.
- Domain events make asynchronous operations easy.
- Domain events provide final consistency across aggregates.
- Domain events can simplify large batch operations into many small business operations.
- Domain events can accomplish powerful event storage.
- Domain events can complete the integration between bounded contexts.
- Domain events are a support for more complex Architectures (CQRS).