EasyNetQ在.Net5中的应用
这篇文章参考网上的教程,主要解决了EasyNetQ的依赖注入服务问题。
主要涉及四个文件:
ApplicationExtenssion.cs .Net中间件的实现
OrderConsumer.cs 消息消费者的实现
WindsorMessageDispatcher.cs 通过容器解析获得消费者服务
Startup.cs 相关服务的注册及中间件引用
主要参考代码:
ApplicationExtenssion.cs
public static class ApplicationExtenssion { ////// 中间件实现 /// /// /// /// /// public static IApplicationBuilder UseSubscribe(this IApplicationBuilder appBuilder, string subscriptionIdPrefix, Assembly assembly) { var services = appBuilder.ApplicationServices.CreateScope().ServiceProvider; var lifeTime = services.GetService (); var bus = services.GetRequiredService (); lifeTime.ApplicationStarted.Register(() => { var subscriber = new AutoSubscriber(bus, subscriptionIdPrefix); //需要指定AutoSubscriberMessageDispatcher对应的实例 //并可以通过构造函数传参,如:IServicesProvider,即:services subscriber.AutoSubscriberMessageDispatcher = new WindsorMessageDispatcher(services); subscriber.Subscribe(new Assembly[] { assembly }); subscriber.SubscribeAsync(new Assembly[] { assembly }); }); lifeTime.ApplicationStopped.Register(() => { bus.Dispose(); }); return appBuilder; } }
OrderConsumer.cs
////// IConsume接口的实现类 /// public class OrderConsumer : IConsume { private readonly ILogger logger; /// /// 可以注入相关服务 /// /// public OrderConsumer(ILogger logger) { this.logger = logger; } /// /// 消息的消费方法 /// /// /// public void Consume(MessageDto message, CancellationToken cancellationToken = default) { logger.LogInformation(JsonConvert.SerializeObject(message)); } }
WindsorMessageDispatcher.cs
////// 结合官方教程实现自动订阅消息转发 /// public class WindsorMessageDispatcher : IAutoSubscriberMessageDispatcher { private readonly IServiceProvider service; /// /// 参数由中间件的AutoSubscriberMessageDispatcher进行传递 /// /// public WindsorMessageDispatcher(IServiceProvider service) { this.service = service; } /// /// 同步方法,可以解析TConsumer对应的服务 /// /// /// /// /// void IAutoSubscriberMessageDispatcher.Dispatch (TMessage message, CancellationToken cancellationToken) { //通过容器获取对应的消费者服务,也可以通过Autofac实现 var consumer = service.GetService >(); //通过解析到的服务调用对应的消费者实现的方法 consumer.Consume(message as MessageDto, cancellationToken); } /// /// 异步方法 /// /// /// /// /// /// async Task IAutoSubscriberMessageDispatcher.DispatchAsync (TMessage message, CancellationToken cancellationToken) { var consumer = service.GetService (); await consumer.ConsumeAsync(message, cancellationToken); } }
Startup.cs
public void ConfigureServices(IServiceCollection services) { string rabbitMqConnection = Configuration["RabbitMqConnection"]; //注册IBus为单例模式 services.AddSingleton(RabbitHutch.CreateBus(rabbitMqConnection)); //注册IConsume的实现类,EasyNetQ会自动扫描实现该接口的类 services.AddSingleton,OrderConsumer>(); services.AddControllers(); services.AddSwaggerGen(c => { c.SwaggerDoc("v1", new OpenApiInfo { Title = "EasyNetQ.WebApi", Version = "v1" }); }); }
public void Configure(IApplicationBuilder app, IWebHostEnvironment env) { if (env.IsDevelopment()) { app.UseDeveloperExceptionPage(); app.UseSwagger(); app.UseSwaggerUI(c => c.SwaggerEndpoint("/swagger/v1/swagger.json", "EasyNetQ.WebApi v1")); } app.UseHttpsRedirection(); app.UseRouting(); app.UseAuthorization(); //添加相应的中间件 app.UseSubscribe("OrderService", Assembly.GetExecutingAssembly()); app.UseEndpoints(endpoints => { endpoints.MapControllers(); }); }
代码:https://github.com/superfeeling-git/EasyNetQ