事件驱动 vs 请求驱动：架构选型决策
在设计微服务通信方式时，工程师常常面临一个核心抉择：使用同步的请求/响应（REST/gRPC），还是异步的事件驱动。两种模式各有适用边界，错误选型会导致系统耦合、性能瓶颈或调试困难。
| 维度 | 请求驱动（REST/gRPC） | 事件驱动（EDA） |
|---|---|---|
| 耦合度 | 调用方知道被调用方 | Producer 不知道 Consumer |
| 可用性 | 下游故障级联影响上游 | 下游故障不影响上游 |
| 响应时间 | 同步等待，尾延迟累积 | 异步处理，用户感觉更快 |
| 一致性 | 强一致性（立刻生效） | 最终一致性（存在窗口期） |
| 调试难度 | 调用栈清晰 | 需要链路追踪 |
| 适用场景 | 实时查询、必须成功 | 通知、解耦、批量处理 |
决策树：
需要同步返回结果？
├─ 是 → 下游必须在本次请求中完成？
│ ├─ 是 → 使用 gRPC / REST 同步调用
│ └─ 否 → 异步 Job + 轮询 / Webhook 回调
└─ 否 → 多个服务需要知道这件事？
├─ 是 → 事件驱动（发布-订阅）
└─ 否 → 消息队列（点对点）
核心原则：不要为了追求"架构先进性"而把所有通信都改为事件驱动。账户余额扣减、库存扣减等强一致性场景， Saga 模式下的补偿事件比纯事件驱动更安全。
事件驱动架构概述
事件驱动架构（EDA）是一种通过事件的产生、检测和消费来实现系统组件间松耦合通信的架构风格。
事件驱动架构优势：
┌─────────────────────────────────────────┐
│ ✓ 松耦合：生产者不知道消费者是谁 │
│ ✓ 可扩展：新增消费者不影响现有系统 │
│ ✓ 审计追踪：完整的事件历史记录 │
│ ✓ 时间解耦：异步处理，提高响应速度 │
│ ✓ 弹性：失败可重试，不影响主流程 │
└─────────────────────────────────────────┘
领域事件设计
事件定义
// src/domain/events/baseEvent.ts
export interface DomainEvent {
eventId: string;
eventType: string;
aggregateId: string;
aggregateType: string;
timestamp: Date;
version: number;
metadata?: Record<string, any>;
}
// src/domain/events/orderEvents.ts
export class OrderCreatedEvent implements DomainEvent {
readonly eventId: string;
readonly eventType = 'OrderCreated';
readonly aggregateType = 'Order';
readonly timestamp: Date;
readonly version = 1;
constructor(
public readonly aggregateId: string,
public readonly userId: string,
public readonly items: OrderItem[],
public readonly totalAmount: number,
public readonly shippingAddress: Address,
metadata?: Record<string, any>
) {
this.eventId = generateUUID();
this.timestamp = new Date();
this.metadata = metadata;
}
}
export class OrderConfirmedEvent implements DomainEvent {
readonly eventId: string;
readonly eventType = 'OrderConfirmed';
readonly aggregateType = 'Order';
readonly timestamp: Date;
readonly version = 1;
constructor(
public readonly aggregateId: string,
public readonly confirmedAt: Date,
public readonly estimatedDelivery: Date,
metadata?: Record<string, any>
) {
this.eventId = generateUUID();
this.timestamp = new Date();
this.metadata = metadata;
}
}
export class OrderCancelledEvent implements DomainEvent {
readonly eventId: string;
readonly eventType = 'OrderCancelled';
readonly aggregateType = 'Order';
readonly timestamp: Date;
readonly version = 1;
constructor(
public readonly aggregateId: string,
public readonly reason: string,
public readonly cancelledBy: string,
metadata?: Record<string, any>
) {
this.eventId = generateUUID();
this.timestamp = new Date();
this.metadata = metadata;
}
}
export class OrderShippedEvent implements DomainEvent {
readonly eventId: string;
readonly eventType = 'OrderShipped';
readonly aggregateType = 'Order';
readonly timestamp: Date;
readonly version = 1;
constructor(
public readonly aggregateId: string,
public readonly trackingNumber: string,
public readonly carrier: string,
public readonly shippedAt: Date,
metadata?: Record<string, any>
) {
this.eventId = generateUUID();
this.timestamp = new Date();
this.metadata = metadata;
}
}
聚合根发布事件
// src/domain/entities/Order.ts
import { DomainEvent } from '../events/baseEvent';
import { OrderCreatedEvent, OrderConfirmedEvent, OrderCancelledEvent } from '../events/orderEvents';
export class Order {
private domainEvents: DomainEvent[] = [];
constructor(
public readonly id: string,
public userId: string,
public items: OrderItem[],
public totalAmount: number,
public shippingAddress: Address,
public status: OrderStatus = 'PENDING',
public createdAt: Date = new Date()
) {}
static create(
userId: string,
items: OrderItem[],
shippingAddress: Address
): Order {
const totalAmount = items.reduce(
(sum, item) => sum + item.price * item.quantity,
0
);
const order = new Order(
generateUUID(),
userId,
items,
totalAmount,
shippingAddress
);
// 发布领域事件
order.addDomainEvent(
new OrderCreatedEvent(
order.id,
userId,
items,
totalAmount,
shippingAddress,
{ correlationId: generateCorrelationId() }
)
);
return order;
}
confirm(estimatedDelivery: Date): void {
if (this.status !== 'PENDING') {
throw new BusinessRuleError('Only pending orders can be confirmed');
}
this.status = 'CONFIRMED';
this.addDomainEvent(
new OrderConfirmedEvent(
this.id,
new Date(),
estimatedDelivery
)
);
}
cancel(reason: string, cancelledBy: string): void {
if (this.status === 'SHIPPED') {
throw new BusinessRuleError('Cannot cancel shipped order');
}
this.status = 'CANCELLED';
this.addDomainEvent(
new OrderCancelledEvent(
this.id,
reason,
cancelledBy
)
);
}
// 获取并清除领域事件
pullDomainEvents(): DomainEvent[] {
const events = [...this.domainEvents];
this.domainEvents = [];
return events;
}
private addDomainEvent(event: DomainEvent): void {
this.domainEvents.push(event);
}
}
// src/application/services/OrderApplicationService.ts
export class OrderApplicationService {
constructor(
private orderRepository: OrderRepository,
private eventPublisher: EventPublisher
) {}
async createOrder(command: CreateOrderCommand): Promise<string> {
// 1. 创建聚合根
const order = Order.create(
command.userId,
command.items,
command.shippingAddress
);
// 2. 保存到数据库（事务）
await this.orderRepository.save(order);
// 3. 发布领域事件
const events = order.pullDomainEvents();
await this.eventPublisher.publishAll(events);
return order.id;
}
async confirmOrder(orderId: string, estimatedDelivery: Date): Promise<void> {
const order = await this.orderRepository.findById(orderId);
if (!order) {
throw new NotFoundError('Order', orderId);
}
order.confirm(estimatedDelivery);
await this.orderRepository.save(order);
const events = order.pullDomainEvents();
await this.eventPublisher.publishAll(events);
}
}
事件总线实现
内存事件总线
// src/infrastructure/events/InMemoryEventBus.ts
import { EventEmitter } from 'events';
import { DomainEvent } from '../../domain/events/baseEvent';
type EventHandler = (event: DomainEvent) => Promise<void>;
export class InMemoryEventBus {
private emitter: EventEmitter;
constructor() {
this.emitter = new EventEmitter();
this.emitter.setMaxListeners(100); // 增加监听器数量限制
}
subscribe(eventType: string, handler: EventHandler): void {
this.emitter.on(eventType, async (event: DomainEvent) => {
try {
await handler(event);
} catch (error) {
console.error(`Error handling event ${eventType}:`, error);
// 可以发送到死信队列
}
});
}
async publish(event: DomainEvent): Promise<void> {
this.emitter.emit(event.eventType, event);
}
async publishAll(events: DomainEvent[]): Promise<void> {
for (const event of events) {
await this.publish(event);
}
}
}
// 使用示例
const eventBus = new InMemoryEventBus();
// 订阅订单创建事件
eventBus.subscribe('OrderCreated', async (event: OrderCreatedEvent) => {
await inventoryService.reserveStock(event.aggregateId, event.items);
});
eventBus.subscribe('OrderCreated', async (event: OrderCreatedEvent) => {
await notificationService.sendOrderConfirmation(event.userId, event.aggregateId);
});
// 订阅订单确认事件
eventBus.subscribe('OrderConfirmed', async (event: OrderConfirmedEvent) => {
await warehouseService.prepareShipment(event.aggregateId);
});
Kafka事件总线
// src/infrastructure/events/KafkaEventBus.ts
import { Kafka, Producer, Consumer, EachMessagePayload } from 'kafkajs';
import { DomainEvent } from '../../domain/events/baseEvent';
export class KafkaEventBus {
private producer: Producer;
private consumers: Map<string, Consumer> = new Map();
private topic: string;
constructor(
private brokers: string[],
private clientId: string,
topic: string = 'domain-events'
) {
const kafka = new Kafka({
clientId,
brokers
});
this.producer = kafka.producer();
this.topic = topic;
}
async connect(): Promise<void> {
await this.producer.connect();
console.log('Kafka producer connected');
}
async publish(event: DomainEvent): Promise<void> {
await this.producer.send({
topic: this.topic,
messages: [
{
key: event.aggregateId,
value: JSON.stringify(event),
headers: {
eventType: event.eventType,
aggregateType: event.aggregateType,
timestamp: event.timestamp.toISOString()
}
}
]
});
console.log(`Published event: ${event.eventType} for ${event.aggregateId}`);
}
async subscribe(
eventType: string,
groupId: string,
handler: (event: DomainEvent) => Promise<void>
): Promise<void> {
const kafka = new Kafka({
clientId: this.clientId,
brokers: this.brokers
});
const consumer = kafka.consumer({ groupId });
await consumer.connect();
await consumer.subscribe({ topic: this.topic, fromBeginning: false });
await consumer.run({
eachMessage: async ({ topic, partition, message }: EachMessagePayload) => {
const eventTypeHeader = message.headers?.eventType?.toString();
if (eventTypeHeader === eventType) {
const event = JSON.parse(message.value?.toString() || '{}');
try {
await handler(event);
} catch (error) {
console.error(`Error handling event ${eventType}:`, error);
// 发送到死信队列或重试
throw error;
}
}
}
});
this.consumers.set(`${eventType}-${groupId}`, consumer);
console.log(`Subscribed to ${eventType} with group ${groupId}`);
}
async disconnect(): Promise<void> {
await this.producer.disconnect();
for (const consumer of this.consumers.values()) {
await consumer.disconnect();
}
console.log('Kafka disconnected');
}
}
// 使用示例
const eventBus = new KafkaEventBus(
['kafka1:9092', 'kafka2:9092'],
'order-service'
);
await eventBus.connect();
// 发布事件
await eventBus.publish(orderCreatedEvent);
// 订阅事件
await eventBus.subscribe(
'OrderCreated',
'inventory-service-group',
async (event) => {
await inventoryService.reserveStock(event.aggregateId, event.items);
}
);
Saga 模式：分布式事务的编排方案
微服务架构下，传统 ACID 事务无法跨服务边界工作。Saga 模式通过一系列本地事务 + 补偿操作，实现最终一致性的"分布式事务"。
Saga 两种实现方式
| 类型 | 原理 | 优点 | 缺点 |
|---|---|---|---|
| 编排式 Saga | 各服务监听事件，自主决定下一步 | 松耦合、弹性好 | 流程分散，难以整体把握 |
| 编排器 Saga | 中央协调器调度各服务 | 流程集中，可视性强 | 协调器成为单点 |
订单处理 Saga（编排式）
// 订单创建流程的 Saga 事件链
// OrderCreatedEvent → InventoryReservedEvent → PaymentProcessedEvent → OrderConfirmedEvent
// 如果 Payment 失败 → PaymentFailedEvent → InventoryReleasedEvent (补偿)
eventBus.subscribe('OrderCreated', 'saga-orchestrator', async (event: OrderCreatedEvent) => {
// 步骤 1：预留库存
await inventoryService.reserveStock(event.aggregateId, event.items);
});
eventBus.subscribe('InventoryReserved', 'saga-orchestrator', async (event) => {
// 步骤 2：扣款
try {
await paymentService.charge(event.orderId, event.totalAmount);
} catch (error) {
// 补偿：释放库存
await eventBus.publish(new InventoryReleasedEvent(event.orderId));
}
});
eventBus.subscribe('PaymentProcessed', 'saga-orchestrator', async (event) => {
// 步骤 3：确认订单
await orderService.confirmOrder(event.orderId);
});
// 补偿事件处理器
eventBus.subscribe('PaymentFailed', 'saga-compensator', async (event) => {
await inventoryService.releaseStock(event.orderId);
await notificationService.sendPaymentFailedEmail(event.userId);
});
Saga 的关键设计点：
- 补偿操作必须幂等：同一补偿事件可能被重试多次
- 补偿顺序与正向操作相反：先做的最后补偿（栈结构）
- 隔离性牺牲：Saga 执行期间，中间状态对外可见（库存已预留但订单未确认）
- 语义锁：通过预留/冻结代替真实扣减，避免客户看到不一致的中间态
对比两阶段提交（2PC）：2PC 是强一致性但阻塞式的（协调器故障会导致资源锁定）；Saga 是最终一致性但非阻塞的（故障只影响当前流程）。互联网系统更倾向于 Saga。
事件溯源（Event Sourcing）
事件存储
// src/infrastructure/persistence/EventStore.ts
import { Pool } from 'pg';
import { DomainEvent } from '../../domain/events/baseEvent';
export class PostgresEventStore {
constructor(private pool: Pool) {
this.initializeSchema();
}
private async initializeSchema(): Promise<void> {
await this.pool.query(`
CREATE TABLE IF NOT EXISTS events (
event_id UUID PRIMARY KEY,
event_type VARCHAR(255) NOT NULL,
aggregate_id UUID NOT NULL,
aggregate_type VARCHAR(255) NOT NULL,
event_data JSONB NOT NULL,
metadata JSONB,
version INTEGER NOT NULL,
created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
UNIQUE(aggregate_id, version)
);
CREATE INDEX IF NOT EXISTS idx_events_aggregate
ON events(aggregate_id, version);
CREATE INDEX IF NOT EXISTS idx_events_type
ON events(event_type, created_at);
`);
}
async saveEvents(
aggregateId: string,
events: DomainEvent[],
expectedVersion: number
): Promise<void> {
const client = await this.pool.connect();
try {
await client.query('BEGIN');
// 乐观锁检查
const currentVersion = await this.getAggregateVersion(client, aggregateId);
if (currentVersion !== expectedVersion) {
throw new ConcurrencyError(
`Expected version ${expectedVersion}, but got ${currentVersion}`
);
}
// 保存事件
for (const event of events) {
await client.query(
`INSERT INTO events
(event_id, event_type, aggregate_id, aggregate_type,
event_data, metadata, version, created_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`,
[
event.eventId,
event.eventType,
event.aggregateId,
event.aggregateType,
JSON.stringify(event),
event.metadata ? JSON.stringify(event.metadata) : null,
event.version,
event.timestamp
]
);
}
await client.query('COMMIT');
} catch (error) {
await client.query('ROLLBACK');
throw error;
} finally {
client.release();
}
}
async loadEvents(aggregateId: string): Promise<DomainEvent[]> {
const result = await this.pool.query(
`SELECT event_data FROM events
WHERE aggregate_id = $1
ORDER BY version ASC`,
[aggregateId]
);
return result.rows.map(row => {
const eventData = row.event_data;
return this.deserializeEvent(eventData);
});
}
private async getAggregateVersion(
client: any,
aggregateId: string
): Promise<number> {
const result = await client.query(
`SELECT MAX(version) as version FROM events WHERE aggregate_id = $1`,
[aggregateId]
);
return result.rows[0]?.version || 0;
}
private deserializeEvent(eventData: any): DomainEvent {
// 根据eventType反序列化为具体的事件类
const eventMap = {
'OrderCreated': OrderCreatedEvent,
'OrderConfirmed': OrderConfirmedEvent,
'OrderCancelled': OrderCancelledEvent,
// ...
};
const EventClass = eventMap[eventData.eventType];
if (!EventClass) {
throw new Error(`Unknown event type: ${eventData.eventType}`);
}
return Object.assign(new EventClass(), eventData);
}
}
// src/domain/aggregates/EventSourcedOrder.ts
export class EventSourcedOrder {
private version: number = 0;
constructor(
public readonly id: string,
public userId: string = '',
public items: OrderItem[] = [],
public totalAmount: number = 0,
public status: OrderStatus = 'PENDING',
private changes: DomainEvent[] = []
) {}
// 从事件流重建状态
loadFromHistory(events: DomainEvent[]): void {
for (const event of events) {
this.apply(event, false);
}
}
// 创建新订单
static create(
userId: string,
items: OrderItem[],
shippingAddress: Address
): EventSourcedOrder {
const order = new EventSourcedOrder(generateUUID());
const totalAmount = items.reduce(
(sum, item) => sum + item.price * item.quantity,
0
);
const event = new OrderCreatedEvent(
order.id,
userId,
items,
totalAmount,
shippingAddress
);
order.apply(event, true);
return order;
}
confirm(estimatedDelivery: Date): void {
if (this.status !== 'PENDING') {
throw new BusinessRuleError('Only pending orders can be confirmed');
}
const event = new OrderConfirmedEvent(
this.id,
new Date(),
estimatedDelivery
);
this.apply(event, true);
}
// 应用事件
private apply(event: DomainEvent, isNew: boolean): void {
switch (event.eventType) {
case 'OrderCreated':
const createdEvent = event as OrderCreatedEvent;
this.userId = createdEvent.userId;
this.items = createdEvent.items;
this.totalAmount = createdEvent.totalAmount;
this.status = 'PENDING';
break;
case 'OrderConfirmed':
this.status = 'CONFIRMED';
break;
case 'OrderCancelled':
this.status = 'CANCELLED';
break;
}
this.version++;
if (isNew) {
this.changes.push(event);
}
}
// 获取未保存的变更
getChanges(): DomainEvent[] {
return this.changes;
}
clearChanges(): void {
this.changes = [];
}
}
// src/application/services/EventSourcedOrderService.ts
export class EventSourcedOrderService {
constructor(
private eventStore: PostgresEventStore,
private eventPublisher: EventPublisher
) {}
async createOrder(command: CreateOrderCommand): Promise<string> {
const order = EventSourcedOrder.create(
command.userId,
command.items,
command.shippingAddress
);
const changes = order.getChanges();
await this.eventStore.saveEvents(order.id, changes, 0);
await this.eventPublisher.publishAll(changes);
order.clearChanges();
return order.id;
}
async confirmOrder(orderId: string, estimatedDelivery: Date): Promise<void> {
const events = await this.eventStore.loadEvents(orderId);
if (events.length === 0) {
throw new NotFoundError('Order', orderId);
}
const order = new EventSourcedOrder(orderId);
order.loadFromHistory(events);
const currentVersion = events.length;
order.confirm(estimatedDelivery);
const changes = order.getChanges();
await this.eventStore.saveEvents(orderId, changes, currentVersion);
await this.eventPublisher.publishAll(changes);
}
}
CQRS（命令查询职责分离）
// src/application/commands/OrderCommands.ts
export interface Command {
commandId: string;
timestamp: Date;
}
export class CreateOrderCommand implements Command {
readonly commandId = generateUUID();
readonly timestamp = new Date();
constructor(
public readonly userId: string,
public readonly items: OrderItem[],
public readonly shippingAddress: Address
) {}
}
// src/application/queries/OrderQueries.ts
export interface OrderReadModel {
id: string;
userId: string;
items: OrderItem[];
totalAmount: number;
status: string;
createdAt: Date;
customerName?: string;
productNames?: string[];
}
export class GetOrderQuery {
constructor(public readonly orderId: string) {}
}
export class GetUserOrdersQuery {
constructor(
public readonly userId: string,
public readonly page: number = 1,
public readonly pageSize: number = 20
) {}
}
// src/infrastructure/persistence/OrderReadModelProjector.ts
export class OrderReadModelProjector {
constructor(private readDb: Pool) {
this.initializeSchema();
}
private async initializeSchema(): Promise<void> {
await this.readDb.query(`
CREATE TABLE IF NOT EXISTS order_read_models (
id UUID PRIMARY KEY,
user_id UUID NOT NULL,
items JSONB NOT NULL,
total_amount DECIMAL NOT NULL,
status VARCHAR(50) NOT NULL,
customer_name VARCHAR(255),
product_names TEXT[],
created_at TIMESTAMP WITH TIME ZONE NOT NULL,
updated_at TIMESTAMP WITH TIME ZONE NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_orders_user
ON order_read_models(user_id, created_at DESC);
CREATE INDEX IF NOT EXISTS idx_orders_status
ON order_read_models(status);
`);
}
async project(event: DomainEvent): Promise<void> {
switch (event.eventType) {
case 'OrderCreated':
await this.handleOrderCreated(event as OrderCreatedEvent);
break;
case 'OrderConfirmed':
await this.updateOrderStatus(event.aggregateId, 'CONFIRMED');
break;
case 'OrderCancelled':
await this.updateOrderStatus(event.aggregateId, 'CANCELLED');
break;
}
}
private async handleOrderCreated(event: OrderCreatedEvent): Promise<void> {
// 查询客户信息
const customer = await this.readDb.query(
'SELECT name FROM users WHERE id = $1',
[event.userId]
);
// 查询产品名称
const productIds = event.items.map(item => item.productId);
const products = await this.readDb.query(
'SELECT id, name FROM products WHERE id = ANY($1)',
[productIds]
);
const productNames = products.rows.map(p => p.name);
await this.readDb.query(
`INSERT INTO order_read_models
(id, user_id, items, total_amount, status, customer_name, product_names, created_at, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $8)`,
[
event.aggregateId,
event.userId,
JSON.stringify(event.items),
event.totalAmount,
'PENDING',
customer.rows[0]?.name || 'Unknown',
productNames,
event.timestamp
]
);
}
private async updateOrderStatus(orderId: string, status: string): Promise<void> {
await this.readDb.query(
`UPDATE order_read_models
SET status = $1, updated_at = NOW()
WHERE id = $2`,
[status, orderId]
);
}
}
// src/application/queries/OrderQueryService.ts
export class OrderQueryService {
constructor(private readDb: Pool) {}
async getOrder(query: GetOrderQuery): Promise<OrderReadModel | null> {
const result = await this.readDb.query(
'SELECT * FROM order_read_models WHERE id = $1',
[query.orderId]
);
return result.rows[0] || null;
}
async getUserOrders(query: GetUserOrdersQuery): Promise<{
orders: OrderReadModel[];
total: number;
}> {
const offset = (query.page - 1) * query.pageSize;
const [ordersResult, countResult] = await Promise.all([
this.readDb.query(
`SELECT * FROM order_read_models
WHERE user_id = $1
ORDER BY created_at DESC
LIMIT $2 OFFSET $3`,
[query.userId, query.pageSize, offset]
),
this.readDb.query(
'SELECT COUNT(*) FROM order_read_models WHERE user_id = $1',
[query.userId]
)
]);
return {
orders: ordersResult.rows,
total: parseInt(countResult.rows[0].count)
};
}
}
// 订阅事件并投影
const projector = new OrderReadModelProjector(readDb);
eventBus.subscribe('OrderCreated', 'read-model-projector', async (event) => {
await projector.project(event);
});
eventBus.subscribe('OrderConfirmed', 'read-model-projector', async (event) => {
await projector.project(event);
});
eventBus.subscribe('OrderCancelled', 'read-model-projector', async (event) => {
await projector.project(event);
});
事件模式反模式与避坑指南
事件驱动架构在实践中容易陷入几个典型陷阱，新手尤其需要注意：
1. 事件贫血症（Anemic Events）
事件只携带 ID，消费者需要再查询一次才能获取完整信息。这导致 N+1 查询问题和额外的服务依赖。正确的做法是：事件应包含足够上下文（至少包含消费者处理所需的核心字段），同时避免过度膨胀。
2. 分布式大泥球（Distributed Big Ball of Mud）
所有服务监听所有事件，事件链无限延伸，一个事件触发十个消费者，每个消费者又发布新事件，形成难以追踪的"事件风暴"。解法是绘制事件依赖图，限制事件传播深度，设置超时熔断。
3. 过早的事件溯源
并非所有实体都适合事件溯源。简单配置表、日志记录等读多写少的数据，直接用关系型存储更经济。事件溯源的适用条件：业务规则复杂、审计要求严格、需要状态回放（如交易、工单、游戏状态）。
4. 消息顺序假设
分布式消息队列不保证全局顺序（除非单分区单消费者）。如果业务依赖严格顺序（如账户流水），需要在应用层实现顺序控制（如基于聚合 ID 的分区或版本号校验）。
5. 忽略死信队列
消费失败后无限重试会阻塞整个分区。必须配置重试策略（如 3 次重试后转入死信队列），并建立死信告警和人工介入机制。
// 带死信队列的 Kafka 消费者配置
const consumer = kafka.consumer({
groupId: 'order-service',
retry: {
retries: 3,
initialRetryTime: 1000,
factor: 2, // 指数退避
},
});
// 消费失败处理
await consumer.run({
eachMessage: async ({ topic, partition, message }) => {
try {
await processMessage(message);
} catch (error) {
// 超过重试次数后发送到死信队列
await deadLetterProducer.send({
topic: `${topic}.dlq`,
messages: [{ key: message.key, value: message.value, headers: { error: error.message } }],
});
}
},
});
总结
事件驱动架构的核心价值：
- 领域事件：捕获业务中发生的重要事实，作为系统通信的基础
- 事件总线：实现组件间的松耦合通信，支持多个消费者
- 事件溯源：通过事件流重建状态，提供完整的审计追踪
- CQRS：分离命令和查询，优化读写性能
- Saga 模式：分布式事务的编排方案，保证最终一致性
关键原则：
- 事件是不可变的事实记录，命名用过去式（OrderCreated，不是 CreateOrder）
- 使用事件实现服务间的松耦合，但不要用事件替代所有同步调用
- 事件溯源提供强大的审计和调试能力，但不是银弹
- CQRS 允许针对读写场景分别优化，但会增加系统复杂度
- 实现幂等性处理重复事件，分布式系统必须假设消息会重复投递
- 死信队列和重试策略是生产环境的底线要求
延伸阅读
- Domain-Driven Design - Eric Evans
- Event Sourcing - Martin Fowler
- CQRS - Martin Fowler
- Apache Kafka文档
- EventStoreDB
- https://plumephp.com/graphql-subscriptions-realtime-updates/ — GraphQL 订阅实现实时事件推送
- https://plumephp.com/websocket-realtime-communication-architecture/ — WebSocket 实时通信架构
- https://plumephp.com/zookeeper-coordination/ — 分布式协调服务
- https://plumephp.com/kafka-connect/ — Kafka Connect 数据集成
- https://plumephp.com/database-replication/ — 数据库主从复制与数据同步
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。