基于Abp VNext框架设计 – Masstransit分布式消息
abp 通过 IDistributedEventBus 接口集成自IEventBus实现分布式事件消息的发布订阅。
IEventBus 在什么时机触发 PublishAsync ?
-
当前UnitOfWork完成时，触发
IEventBus的PublishAsync -
在没有事务环境下，同步调用
IEventBus的PublishAsync
abp 默认实现基于RabbitMq消息队列 Volo.Abp.EventBus.RabbitMQ 实现分布式消息的发布与订阅。
消息治理核心问题：
-
生产端如何保证投递成功的消息不能丢失。
-
Mq自身如何保证消息不丢失。
-
消费段如何保证消费端的消息不丢失。
基于abp 默认实现的DistributedEventBus不能满足以下场景：
-
Publisher 生产者无法保证消息一定能投递到MQ。
-
Consumer 消费端在消息消费时，出现异常时，没有异常错误处理机制（确保消费失败的消息能重新被消费）。
我们引入Masstransit，来提升abp对消息治理能力。
Masstransit提供以下开箱即用功能：
-
Publish/Send/Request-Response等几种消息投递机制。
-
多种IOC容器支持。
-
异常机制。
-
Saga事务管理。
-
事务活动补偿机制（Courier）
-
消息审计
-
消息管道处理机制
Abp 框架下事件消息集成
-
使用MassTransit重新实现
IDistributedEventBus。 -
在消费端Consumer传递用户身份信息。
-
使用Asp.Net Core Web Host 作消费端Consumer宿主。
集成MassTransit
在Module初始化时，注入MassTransit实例，并启动。
Copy
///
/// 配置DistributedEventBus
///
///
///
///
private void ConfigureDistributedEventBus(ServiceConfigurationContext context, IConfiguration configuration, IWebHostEnvironment hostingEnvironment)
{
var options = context.Services.GetConfiguration().GetSection("Rabbitmq").Get();
var mqConnectionString = "rabbitmq://" + options.ConnectionString;
context.Services.AddMassTransit(mtConfig =>
{
//inject consumers into IOC from assembly
mtConfig.AddConsumers(typeof(AuthCenterEventBusHostModule));
mtConfig.AddBus(provider =>
{
var bus = Bus.Factory.CreateUsingRabbitMq(mqConfig =>
{
var host = mqConfig.Host(new Uri(mqConnectionString), h =>
{
h.Username(options.UserName);
h.Password(options.Password);
});
// set special message serializer
mqConfig.UseBsonSerializer();
// integrated existed logger compontent
mqConfig.UseExtensionsLogging(provider.GetService());
mqConfig.ReceiveEndpoint(host, "authcenter-queue", q =>
{
//set rabbitmq prefetch count
q.PrefetchCount = 200;
//set message retry policy
q.UseMessageRetry(r => r.Interval(3, 100));
q.Consumer(provider);
EndpointConvention.Map(q.InputAddress);
});
mqConfig.ReceiveEndpoint(host, "user-synchronization", q =>
{
//set rabbitmq prefetch count
q.PrefetchCount = 50;
//q.UseRateLimit(100, TimeSpan.FromSeconds(1));
//q.UseConcurrencyLimit(2);
//set message retry policy
q.UseMessageRetry(r => r.Interval(3, 100));
q.Consumer(provider);
EndpointConvention.Map(q.InputAddress);
});
mqConfig.ConfigureEndpoints(provider);
mqConfig.UseAuditingFilter(provider, o =>
{
o.ReplaceAuditing = true;
});
});
// set authtication middleware for user identity
bus.ConnectAuthenticationObservers(provider);
return bus;
});
});
}
在MassTransit中，使用 IBusControl 接口 StartAsync 或 StopAsync 来启动或停止。
使用 IPublishEndpoint 重新实现 IDistributedEventBus 接口，实现与abp分布式事件总线集成。
Copy
public class MassTransitDistributedEventBus : IDistributedEventBus, ISingletonDependency
{
private readonly IPublishEndpoint _publishEndpoint;
//protected IHybridServiceScopeFactory ServiceScopeFactory { get; }
protected AbpDistributedEventBusOptions DistributedEventBusOptions { get; }
public MassTransitDistributedEventBus(
IOptions distributedEventBusOptions,
IPublishEndpoint publishEndpoint)
{
//ServiceScopeFactory = serviceScopeFactory;
_publishEndpoint = publishEndpoint;
DistributedEventBusOptions = distributedEventBusOptions.Value;
//Subscribe(distributedEventBusOptions.Value.Handlers);
}
/*
* Not Implementation
*/
public Task PublishAsync(TEvent eventData)
where TEvent : class
{
return _publishEndpoint.Publish(eventData);
}
public Task PublishAsync(Type eventType, object eventData)
{
return _publishEndpoint.Publish(eventData, eventType);
}
}
到此，我们实现了MassTransit与Abp集成。
事件消息传递User Claims
在实际业务实现过程中，我们会用消息队列实现“削峰填谷”的效果。异步消息队列中传递用户身份信息如何实现呢？
我们先看看abp在WebApi中，如何确定当前用户？
ICurrentUser 提供当前User Claims抽象。而 ICurrentUser 依赖于 ICurrentPrincipalAccessor ，在Asp.Net core中利用HttpContext User 来记录当前用户身份。
在MassTransit中，利用 IPublishObserver > IConsumeObserver 生产者/消费端的观察者，来实现传递已认证的用户Claims。
Copy
///
/// 生产者传递当前用户Principal
///
public class AuthPublishObserver : IPublishObserver
{
private readonly ICurrentPrincipalAccessor _currentPrincipalAccessor;
private readonly IClaimsPrincipalFactory _claimsPrincipalFactory;
public AuthPublishObserver(
ICurrentPrincipalAccessor currentPrincipalAccessor,
IClaimsPrincipalFactory claimsPrincipalFactory)
{
_currentPrincipalAccessor = currentPrincipalAccessor;
_claimsPrincipalFactory = claimsPrincipalFactory;
}
public Task PrePublish(PublishContext context) where T : class
{
var claimsPrincipal = _claimsPrincipalFactory
.CreateClaimsPrincipal(
_currentPrincipalAccessor.Principal
);
if (claimsPrincipal != null)
{
context.Headers.SetAuthenticationHeaders(claimsPrincipal);
}
return TaskUtil.Completed;
}
public Task PostPublish(PublishContext context) where T : class => TaskUtil.Completed;
public Task PublishFault(PublishContext context, Exception exception) where T : class => TaskUtil.Completed;
}
Copy
///
/// 消费端从MqMessage Heads 中获取当前用户Principal，并赋值给HttpContext
///
public class AuthConsumeObserver : IConsumeObserver
{
private readonly IHttpContextAccessor _httpContextAccessor;
private readonly IServiceScopeFactory _factory;
public AuthConsumeObserver(IHttpContextAccessor httpContextAccessor, IServiceScopeFactory factory)
{
_httpContextAccessor = httpContextAccessor;
_factory = factory;
}
public Task PreConsume(ConsumeContext context) where T : class
{
if (_httpContextAccessor.HttpContext == null)
{
_httpContextAccessor.HttpContext = new DefaultHttpContext
{
RequestServices = _factory.CreateScope().ServiceProvider
};
}
var abpClaimsPrincipal = context.Headers.TryGetAbpClaimsPrincipal();
if (abpClaimsPrincipal != null && abpClaimsPrincipal.IsAuthenticated)
{
var claimsPrincipal = abpClaimsPrincipal.ToClaimsPrincipal();
_httpContextAccessor.HttpContext.User = claimsPrincipal;
Thread.CurrentPrincipal = claimsPrincipal;
}
return TaskUtil.Completed;
}
public Task PostConsume(ConsumeContext context) where T : class
{
_httpContextAccessor.HttpContext = null;
return TaskUtil.Completed;
}
public Task ConsumeFault(ConsumeContext context, Exception exception) where T : class
{
_httpContextAccessor.HttpContext = null;
return TaskUtil.Completed;
}
}
使用Asp.Net Core Web Host 作消费端Consumer宿主
基于以下几点原因，我们使用Asp.Net Core Web Host 作为消息端Consumer宿主
-
部署在Linux环境下，Asp.Net Core Web Host 通常使用守护进程来启动服务实例，这样可以保证服务不被中断。
-
根据abp vnext DDD 项目分层，最大程度利用Application层应用方法，复用abp vnext 框架机制。
MassTransit 深入研究
-
延迟消息
-
限流、熔断降级
-
批量消费
-
Saga
References
-
abp vnext disctributed event bus
-
MassTransit
About The Author
