Event-driven architecture enables loose coupling between microservices. Services communicate through events rather than direct calls, improving scalability, resilience, and flexibility. Here's how to build event-driven systems effectively.
Why Events?
Request/Response vs Events
Before adopting event-driven architecture, it's important to understand what problems it solves. Traditional request/response patterns create tight coupling and cascading failures, while events enable independent operation. The following diagram illustrates the fundamental difference between these two approaches, showing how synchronous calls create dependencies while events allow services to operate independently.
Synchronous (Request/Response):
Order Service → User Service → Payment Service → Inventory Service
↓ ↓ ↓
Wait... Wait... Wait...
Problems:
- Tight coupling
- Cascading failures
- Latency accumulation
- Difficult to scale
Event-Driven:
Order Service → publishes "OrderCreated"
├── User Service listens → updates user stats
├── Payment Service listens → processes payment
├── Inventory Service listens → reserves stock
└── Notification Service listens → sends email
Benefits:
- Loose coupling
- Independent scaling
- Resilience to failures
- Easier to add new consumers
With events, each service operates independently. If the notification service is down, orders still process. Adding a new analytics service requires no changes to existing services.
Event Types
Domain Events
Domain events represent something meaningful that happened within your business domain. They should be named in past tense and describe what occurred, not what you want to happen. When you design your domain events, think about the business significance of each occurrence and what information consumers will need.
// Something that happened in the domain
class OrderPlaced implements DomainEvent
{
public function __construct(
public readonly string $orderId,
public readonly string $customerId,
public readonly array $items,
public readonly Money $total,
public readonly DateTimeImmutable $occurredAt,
) {}
}
// Naming: Past tense, describes what happened
// OrderPlaced (not CreateOrder)
// PaymentReceived (not ProcessPayment)
// InventoryReserved (not ReserveInventory)
You'll notice the event uses readonly properties and immutable types. Events represent facts that have already happened, so they should never be modified after creation.
Integration Events
Integration events are designed for communication between bounded contexts. They contain only the information that external services need, avoiding leaking internal implementation details. This separation is crucial for maintaining loose coupling between services.
// Events shared between bounded contexts
class OrderPlacedIntegrationEvent
{
public function __construct(
public readonly string $eventId,
public readonly string $eventType,
public readonly DateTimeImmutable $timestamp,
public readonly array $payload,
) {}
public static function fromDomainEvent(OrderPlaced $event): self
{
return new self(
eventId: Uuid::uuid4()->toString(),
eventType: 'order.placed',
timestamp: $event->occurredAt,
payload: [
'order_id' => $event->orderId,
'customer_id' => $event->customerId,
'total' => [
'amount' => $event->total->getAmount(),
'currency' => $event->total->getCurrency(),
],
],
);
}
}
The transformation from domain event to integration event is intentional. It creates a stable contract that other services can depend on, independent of your internal domain model.
Event Schema
A well-designed event schema includes metadata for debugging, tracing, and versioning. This structure has proven useful across many event-driven systems. You can adapt this schema to your specific needs, but ensure you include the essential fields for tracing and versioning.
{
"eventId": "550e8400-e29b-41d4-a716-446655440000",
"eventType": "order.placed",
"aggregateId": "order-123",
"aggregateType": "Order",
"timestamp": "2024-01-15T10:30:00Z",
"version": 1,
"metadata": {
"correlationId": "req-456",
"causationId": "cmd-789",
"userId": "user-123"
},
"payload": {
"orderId": "order-123",
"customerId": "customer-456",
"items": [...],
"total": {"amount": "99.99", "currency": "USD"}
}
}
The correlationId links all events in a single business transaction, while causationId identifies which event or command caused this event. Both are invaluable for debugging distributed systems.
Event Bus Implementation
Simple In-Memory Bus
For testing or single-process applications, an in-memory event bus provides a simple starting point. You can later swap in a distributed implementation without changing your application code. This pattern follows the dependency inversion principle, keeping your business logic decoupled from infrastructure concerns.
class EventBus
{
private array $listeners = [];
public function subscribe(string $eventType, callable $listener): void
{
$this->listeners[$eventType][] = $listener;
}
public function publish(object $event): void
{
$eventType = get_class($event);
foreach ($this->listeners[$eventType] ?? [] as $listener) {
$listener($event);
}
}
}
// Usage
$bus = new EventBus();
$bus->subscribe(OrderPlaced::class, function (OrderPlaced $event) {
// Handle event
});
$bus->publish(new OrderPlaced(...));
While this implementation is synchronous and in-memory, it establishes the pattern that your application will use. When you switch to Kafka or RabbitMQ, only the bus implementation changes.
Kafka Producer
For distributed systems, Kafka provides durable, high-throughput event streaming. The producer sends events to a topic, where they're persisted and made available to consumers. You'll want to configure acknowledgment settings and retries based on your durability requirements.
class KafkaEventPublisher
{
private Producer $producer;
private string $topic;
public function publish(IntegrationEvent $event): void
{
$message = json_encode([
'event_id' => $event->eventId,
'event_type' => $event->eventType,
'timestamp' => $event->timestamp->format('c'),
'payload' => $event->payload,
]);
$this->producer->produce($this->topic, $event->aggregateId, $message);
$this->producer->flush(10000);
}
}
Using the aggregateId as the partition key ensures all events for the same entity go to the same partition, maintaining ordering guarantees.
Kafka Consumer
The consumer polls for new messages, processes them, and commits offsets to track progress. Error handling is crucial here since failed messages can block the queue. You should carefully consider your error handling strategy before deploying to production.
class KafkaEventConsumer
{
private Consumer $consumer;
private EventHandlerRegistry $handlers;
public function consume(): void
{
$this->consumer->subscribe(['orders']);
while (true) {
$message = $this->consumer->consume(1000);
if ($message->err === RD_KAFKA_RESP_ERR_NO_ERROR) {
$this->handleMessage($message);
$this->consumer->commit($message);
}
}
}
private function handleMessage($message): void
{
$event = json_decode($message->payload, true);
$handler = $this->handlers->getHandler($event['event_type']);
try {
$handler->handle($event);
} catch (Exception $e) {
$this->handleFailure($message, $e);
}
}
}
Only commit the offset after successful processing. If processing fails, the message will be redelivered on restart.
Patterns
Event Sourcing
Event sourcing stores all changes as a sequence of events rather than just the current state. This provides a complete audit trail and enables powerful features like temporal queries and replay. When you implement event sourcing, you're fundamentally changing how your application stores data.
// Store events as source of truth
class OrderAggregate
{
private string $id;
private OrderStatus $status;
private array $items = [];
private array $uncommittedEvents = [];
public static function place(string $id, string $customerId, array $items): self
{
$order = new self();
$order->apply(new OrderPlaced($id, $customerId, $items, new DateTimeImmutable()));
return $order;
}
public function confirm(): void
{
if ($this->status !== OrderStatus::Pending) {
throw new InvalidOrderStateException();
}
$this->apply(new OrderConfirmed($this->id, new DateTimeImmutable()));
}
private function apply(DomainEvent $event): void
{
$this->when($event);
$this->uncommittedEvents[] = $event;
}
private function when(DomainEvent $event): void
{
match (get_class($event)) {
OrderPlaced::class => $this->whenOrderPlaced($event),
OrderConfirmed::class => $this->whenOrderConfirmed($event),
};
}
private function whenOrderPlaced(OrderPlaced $event): void
{
$this->id = $event->orderId;
$this->status = OrderStatus::Pending;
$this->items = $event->items;
}
private function whenOrderConfirmed(OrderConfirmed $event): void
{
$this->status = OrderStatus::Confirmed;
}
public function getUncommittedEvents(): array
{
return $this->uncommittedEvents;
}
}
Notice the separation between the apply method (which records the event) and the when methods (which update state). This separation enables event replay.
CQRS (Command Query Responsibility Segregation)
CQRS separates write operations from read operations, allowing you to optimize each independently. The write model handles commands and produces events, while the read model consumes events and updates query-optimized views. This pattern pairs naturally with event sourcing.
// Write side - handles commands, produces events
class PlaceOrderHandler
{
public function handle(PlaceOrderCommand $command): void
{
$order = OrderAggregate::place(
$command->orderId,
$command->customerId,
$command->items
);
$this->eventStore->save($order);
}
}
// Read side - consumes events, updates query models
class OrderProjection
{
public function handleOrderPlaced(OrderPlaced $event): void
{
DB::table('order_read_model')->insert([
'id' => $event->orderId,
'customer_id' => $event->customerId,
'status' => 'pending',
'item_count' => count($event->items),
'created_at' => $event->occurredAt,
]);
}
public function handleOrderConfirmed(OrderConfirmed $event): void
{
DB::table('order_read_model')
->where('id', $event->orderId)
->update([
'status' => 'confirmed',
'confirmed_at' => $event->occurredAt,
]);
}
}
// Query side - reads from optimized read model
class OrderQuery
{
public function getOrdersByCustomer(string $customerId): array
{
return DB::table('order_read_model')
->where('customer_id', $customerId)
->orderBy('created_at', 'desc')
->get();
}
}
The read model can be denormalized and optimized for specific query patterns without affecting the write model's integrity.
Saga Pattern
Sagas coordinate distributed transactions across multiple services. Each step either succeeds or triggers compensating actions to undo previous steps. Understanding sagas is essential when you need to maintain consistency across service boundaries without distributed transactions.
// Coordinate transactions across services
class OrderSaga
{
private array $completedSteps = [];
public function handle(OrderPlaced $event): void
{
try {
// Step 1: Reserve inventory
$this->inventoryService->reserve($event->orderId, $event->items);
$this->completedSteps[] = 'inventory_reserved';
// Step 2: Process payment
$this->paymentService->charge($event->orderId, $event->total);
$this->completedSteps[] = 'payment_processed';
// Step 3: Confirm order
$this->orderService->confirm($event->orderId);
$this->completedSteps[] = 'order_confirmed';
} catch (Exception $e) {
$this->compensate($event);
throw $e;
}
}
private function compensate(OrderPlaced $event): void
{
// Reverse completed steps in reverse order
foreach (array_reverse($this->completedSteps) as $step) {
match ($step) {
'payment_processed' => $this->paymentService->refund($event->orderId),
'inventory_reserved' => $this->inventoryService->release($event->orderId),
'order_confirmed' => $this->orderService->cancel($event->orderId),
};
}
}
}
// Event-driven saga (choreography)
class InventoryService
{
public function handleOrderPlaced(OrderPlaced $event): void
{
try {
$this->reserveStock($event->items);
$this->eventBus->publish(new InventoryReserved($event->orderId));
} catch (Exception $e) {
$this->eventBus->publish(new InventoryReservationFailed($event->orderId, $e->getMessage()));
}
}
}
The choreography approach (event-driven saga) is more loosely coupled but harder to follow. The orchestration approach (centralized saga) is easier to understand but creates a coordination point.
Eventual Consistency
Handling Eventual Consistency
In event-driven systems, data is eventually consistent, not immediately consistent. The read model may lag behind the write model by milliseconds to seconds. Your application must handle this gracefully, and you should design your user interface to set appropriate expectations.
// Problem: Read immediately after write may return stale data
class OrderController
{
public function createOrder(Request $request): Response
{
// Command published, but read model not updated yet
$this->commandBus->dispatch(new PlaceOrderCommand($request->all()));
// This might return stale data!
$order = $this->orderQuery->find($request->orderId);
return response()->json($order);
}
}
// Solution 1: Return command result, not query
public function createOrder(Request $request): Response
{
$orderId = Uuid::uuid4()->toString();
$this->commandBus->dispatch(new PlaceOrderCommand($orderId, $request->all()));
return response()->json([
'order_id' => $orderId,
'status' => 'processing',
], 202); // Accepted, not OK
}
// Solution 2: Wait for projection
public function createOrder(Request $request): Response
{
$orderId = Uuid::uuid4()->toString();
$this->commandBus->dispatch(new PlaceOrderCommand($orderId, $request->all()));
// Poll until read model updated (with timeout)
$order = retry(5, function () use ($orderId) {
return $this->orderQuery->find($orderId)
?? throw new Exception('Not ready');
}, 100);
return response()->json($order);
}
The HTTP 202 (Accepted) status code is ideal for these scenarios, indicating that the request was accepted for processing but hasn't completed yet.
Idempotency
Idempotent Consumers
Events may be delivered more than once due to retries or consumer restarts. Your handlers must be idempotent, producing the same result whether called once or multiple times. This is not optional in production systems; you must design for at-least-once delivery.
class IdempotentEventHandler
{
private ProcessedEventStore $processedEvents;
public function handle(array $event): void
{
$eventId = $event['event_id'];
// Check if already processed
if ($this->processedEvents->exists($eventId)) {
Log::info("Event {$eventId} already processed, skipping");
return;
}
// Process event
DB::transaction(function () use ($event, $eventId) {
$this->doHandle($event);
// Mark as processed in same transaction
$this->processedEvents->save($eventId);
});
}
}
// Using natural idempotency keys
class PaymentHandler
{
public function handleOrderPlaced(OrderPlaced $event): void
{
// Use orderId as idempotency key
$existingPayment = Payment::where('order_id', $event->orderId)->first();
if ($existingPayment) {
return; // Already processed
}
Payment::create([
'order_id' => $event->orderId,
'amount' => $event->total,
]);
}
}
Storing the processed event ID in the same transaction as the side effects ensures atomicity. If the transaction fails, both the processing and the marker are rolled back.
Dead Letter Queue
When an event cannot be processed after several retries, it should be moved to a dead letter queue for manual investigation rather than blocking the queue indefinitely. This pattern prevents a single bad message from stopping all event processing.
class EventConsumer
{
private const MAX_RETRIES = 3;
public function consume(Message $message): void
{
$retryCount = $message->getHeader('retry_count') ?? 0;
try {
$this->handler->handle($message);
} catch (Exception $e) {
if ($retryCount < self::MAX_RETRIES) {
// Retry with backoff
$this->retryQueue->publish(
$message->withHeader('retry_count', $retryCount + 1),
delay: pow(2, $retryCount) * 1000 // Exponential backoff
);
} else {
// Send to DLQ for manual review
$this->deadLetterQueue->publish($message, [
'error' => $e->getMessage(),
'trace' => $e->getTraceAsString(),
'failed_at' => now()->toIso8601String(),
]);
}
}
}
}
Include enough context in the dead letter message to debug the issue without accessing logs. The error message, stack trace, and timestamp are essential.
Monitoring
Monitoring event processing is crucial for maintaining system health. Track both success/failure counts and processing latency to identify problems early. You should set up alerts for anomalies in these metrics.
// Track event processing metrics
class MonitoredEventHandler
{
public function handle(array $event): void
{
$startTime = microtime(true);
try {
$this->inner->handle($event);
$this->metrics->increment('events.processed', [
'type' => $event['event_type'],
'status' => 'success',
]);
} catch (Exception $e) {
$this->metrics->increment('events.processed', [
'type' => $event['event_type'],
'status' => 'failure',
]);
throw $e;
} finally {
$duration = microtime(true) - $startTime;
$this->metrics->histogram('events.processing_time', $duration, [
'type' => $event['event_type'],
]);
}
}
}
Alert on unusual patterns: high failure rates, increasing processing times, or growing queue depth. These are early indicators of problems.
Conclusion
Event-driven microservices enable loose coupling and independent scaling. Use domain events within bounded contexts and integration events between them. Implement idempotent consumers to handle duplicate events safely. Use sagas for distributed transactions with compensation logic. Accept eventual consistency and design UIs accordingly. Monitor event processing metrics and use dead letter queues for failed events. The complexity is worthwhile for systems that need resilience and scalability.
