一、事件总线在领域驱动设计中的核心作用
在领域驱动设计(DDD)中，事件总线就像是一个超级快递员，负责在不同领域之间传递重要消息。想象一下，当用户在电商平台下单时，订单服务需要通知库存服务、支付服务和物流服务，这时候事件总线就能大显身手了。
使用C#和.NET Core技术栈，我们可以这样实现一个简单的事件发布：
// 定义领域事件接口
public interface IDomainEvent
{
DateTime OccurredOn { get; }
}
// 订单创建事件实现
public class OrderCreatedEvent : IDomainEvent
{
public Guid OrderId { get; }
public decimal Amount { get; }
public DateTime OccurredOn { get; } = DateTime.UtcNow;
public OrderCreatedEvent(Guid orderId, decimal amount)
{
OrderId = orderId;
Amount = amount;
}
}
// 事件发布者
public class EventPublisher
{
private readonly IEventBus _eventBus;
public EventPublisher(IEventBus eventBus)
{
_eventBus = eventBus;
}
public void CreateOrder(Order order)
{
// 业务逻辑...
// 发布事件
var @event = new OrderCreatedEvent(order.Id, order.TotalAmount);
_eventBus.Publish(@event);
}
}
事件总线最大的好处就是解耦。各个服务不需要知道彼此的存在，只需要关心自己感兴趣的事件。这种设计让系统更容易扩展和维护，特别是在微服务架构中特别有用。
二、自研事件总线的实现路径
自己造轮子听起来很酷，但真的适合你吗？让我们看看自研事件总线需要考虑哪些方面。
首先，我们需要一个基本的事件总线实现。还是用C#/.NET Core举例：
// 事件总线接口
public interface IEventBus
{
void Subscribe<TEvent>(Action<TEvent> handler) where TEvent : IDomainEvent;
void Publish<TEvent>(TEvent @event) where TEvent : IDomainEvent;
}
// 内存事件总线实现
public class InMemoryEventBus : IEventBus
{
private readonly Dictionary<Type, List<Delegate>> _handlers = new();
public void Subscribe<TEvent>(Action<TEvent> handler) where TEvent : IDomainEvent
{
var eventType = typeof(TEvent);
if (!_handlers.ContainsKey(eventType))
{
_handlers[eventType] = new List<Delegate>();
}
_handlers[eventType].Add(handler);
}
public void Publish<TEvent>(TEvent @event) where TEvent : IDomainEvent
{
var eventType = typeof(TEvent);
if (_handlers.TryGetValue(eventType, out var handlers))
{
foreach (var handler in handlers)
{
// 同步调用处理程序
((Action<TEvent>)handler)(@event);
}
}
}
}
自研的优势很明显：完全可控，可以根据业务需求定制功能。比如你可以轻松添加重试机制、死信队列等特性。但缺点也很明显：需要投入大量开发资源，而且容易踩坑。
这里有个进阶版，增加了异步处理和错误处理：
public class AdvancedEventBus : IEventBus
{
// 省略部分代码...
public async Task PublishAsync<TEvent>(TEvent @event) where TEvent : IDomainEvent
{
var eventType = typeof(TEvent);
if (_handlers.TryGetValue(eventType, out var handlers))
{
var tasks = new List<Task>();
foreach (var handler in handlers)
{
try
{
// 异步执行处理程序
var task = Task.Run(() => ((Action<TEvent>)handler)(@event));
tasks.Add(task);
}
catch (Exception ex)
{
// 错误处理逻辑
await _deadLetterQueue.AddAsync(@event, ex);
}
}
await Task.WhenAll(tasks);
}
}
}
自研事件总线适合的场景：业务非常特殊，现有框架无法满足需求；团队技术实力雄厚；对性能有极致要求。
三、主流第三方事件总线框架对比
既然自研这么麻烦，为什么不直接用现成的呢？让我们看看几个主流的选择。
- MediatR - .NET生态中的轻量级选择
// 使用MediatR发布事件
public class OrderService
{
private readonly IMediator _mediator;
public OrderService(IMediator mediator)
{
_mediator = mediator;
}
public async Task CreateOrderAsync(Order order)
{
// 业务逻辑...
// 发布事件
await _mediator.Publish(new OrderCreatedEvent(order.Id, order.TotalAmount));
}
}
// 事件处理器
public class OrderCreatedEventHandler : INotificationHandler<OrderCreatedEvent>
{
public async Task Handle(OrderCreatedEvent notification, CancellationToken cancellationToken)
{
// 处理订单创建事件
await _inventoryService.ReduceStockAsync(notification.OrderId);
}
}
MediatR的优点：简单易用，与ASP.NET Core集成好。缺点：功能相对简单，不适合复杂场景。
- MassTransit - 企业级解决方案
// 配置MassTransit
services.AddMassTransit(x =>
{
x.UsingRabbitMq((context, cfg) =>
{
cfg.Host("localhost", "/", h =>
{
h.Username("guest");
h.Password("guest");
});
cfg.ReceiveEndpoint("order-created", e =>
{
e.Consumer<OrderCreatedConsumer>();
});
});
});
// 消费者实现
public class OrderCreatedConsumer : IConsumer<OrderCreatedEvent>
{
public async Task Consume(ConsumeContext<OrderCreatedEvent> context)
{
var message = context.Message;
await _paymentService.ProcessPaymentAsync(message.OrderId, message.Amount);
}
}
MassTransit的优势：支持多种传输方式(RabbitMQ, Azure Service Bus等)，功能强大。缺点：学习曲线较陡，配置复杂。
- Kafka - 大数据场景的首选
虽然Kafka严格来说不是专门的事件总线框架，但在高吞吐量场景下表现出色。
// 生产者配置
var config = new ProducerConfig { BootstrapServers = "localhost:9092" };
// 发布事件
using var producer = new ProducerBuilder<Null, string>(config).Build();
var evt = new OrderCreatedEvent(orderId, amount);
var message = new Message<Null, string>
{
Value = JsonSerializer.Serialize(evt)
};
await producer.ProduceAsync("order-events", message);
Kafka的优势：超高吞吐量，持久化存储，支持流处理。缺点：运维复杂，资源消耗大。
四、选型决策的关键因素
面对这么多选择，到底该怎么选？这里有几个关键考量点：
性能需求：如果你的系统每秒要处理成千上万的事件，Kafka或RabbitMQ+MassTransit可能是更好的选择。如果是普通业务系统，MediatR就够用了。
可靠性要求：金融级系统需要确保事件不丢失，这时候就要选择支持持久化和事务的消息队列。
团队熟悉度：如果团队已经很熟悉RabbitMQ，那么选择MassTransit会比从零学习Kafka更高效。
运维能力：Kafka需要专业的运维知识，小团队可能更适合使用托管服务或更简单的解决方案。
未来扩展：考虑系统未来3-5年的发展，选择能够支撑业务增长的技术。
这里有一个决策树代码示例：
public IEventBus RecommendEventBus(Requirements requirements)
{
if (requirements.HighThroughput)
{
return requirements.HasDevOpsTeam
? new KafkaEventBus()
: new ManagedKafkaEventBus();
}
if (requirements.StrongConsistency)
{
return new MassTransitWithRabbitMQ();
}
if (requirements.SimpleIntegration)
{
return new MediatREventBus();
}
return new InMemoryEventBus(); // 默认选择
}
五、实施注意事项与最佳实践
不管你选择哪种方案，这些经验都值得参考：
事件设计原则：
- 事件命名使用过去时态，如OrderCreated
- 包含足够的信息，但不要太多
- 保持事件的不可变性
错误处理：
- 实现死信队列存储失败的消息
- 考虑重试策略，比如指数退避
监控：
- 记录事件处理耗时
- 监控积压消息数量
- 设置告警机制
这里有一个增强版的事件总线实现，加入了监控和重试：
public class MonitoredEventBus : IEventBus
{
private readonly IEventBus _innerBus;
private readonly IMonitorService _monitor;
public MonitoredEventBus(IEventBus innerBus, IMonitorService monitor)
{
_innerBus = innerBus;
_monitor = monitor;
}
public async Task PublishAsync<TEvent>(TEvent @event) where TEvent : IDomainEvent
{
var stopwatch = Stopwatch.StartNew();
try
{
await _innerBus.PublishAsync(@event);
_monitor.RecordSuccess(typeof(TEvent).Name, stopwatch.ElapsedMilliseconds);
}
catch (Exception ex)
{
_monitor.RecordFailure(typeof(TEvent).Name, ex);
// 重试逻辑
await RetryPolicy
.Handle<Exception>()
.WaitAndRetryAsync(3, retryAttempt =>
TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)))
.ExecuteAsync(() => _innerBus.PublishAsync(@event));
}
}
}
六、总结与个人建议
经过这么多分析，我的建议是：
对于大多数中小型项目，MediatR+内存总线已经足够好用了，简单就是美。
当需要跨服务通信时，MassTransit+RabbitMQ是个平衡的选择，既有足够的功能，又不会太复杂。
只有当你真的需要处理海量事件(比如每秒上万)时，才应该考虑Kafka。
自研只适合两种情况：现有方案完全无法满足需求，或者自研本身就是你们的核心竞争力。
记住，没有最好的方案，只有最适合的方案。选择时要综合考虑团队能力、业务需求和未来发展。
评论