EasyNetQ在.Net5中的應用

這篇文章參考網上的教程，主要解決了EasyNetQ的依賴注入服務問題，

主要涉及四個檔案：

ApplicationExtenssion.cs　　　　.Net中間件的實作

OrderConsumer.cs　　　　   　　訊息消費者的實作

WindsorMessageDispatcher.cs　  通過容器決議獲得消費者服務

Startup.cs　　　　　　　　　　   相關服務的注冊及中間件參考

主要參考代碼：

ApplicationExtenssion.cs

    public static class ApplicationExtenssion
    {
        /// <summary>
        /// 中間件實作
        /// </summary>
        /// <param name="appBuilder"></param>
        /// <param name="subscriptionIdPrefix"></param>
        /// <param name="assembly"></param>
        /// <returns></returns>
        public static IApplicationBuilder UseSubscribe(this IApplicationBuilder appBuilder, string subscriptionIdPrefix, Assembly assembly)
        {
            var services = appBuilder.ApplicationServices.CreateScope().ServiceProvider;
            var lifeTime = services.GetService<IHostApplicationLifetime>();
            var bus = services.GetRequiredService<IBus>();
            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

    /// <summary>
    /// IConsume介面的實作類
    /// </summary>
    public class OrderConsumer : IConsume<MessageDto>
    {
        private readonly ILogger<OrderConsumer> logger;

        /// <summary>
        /// 可以注入相關服務
        /// </summary>
        /// <param name="logger"></param>
        public OrderConsumer(ILogger<OrderConsumer> logger)
        {
            this.logger = logger;
        }

        /// <summary>
        /// 訊息的消費方法
        /// </summary>
        /// <param name="message"></param>
        /// <param name="cancellationToken"></param>
        public void Consume(MessageDto message, CancellationToken cancellationToken = default)
        {
            logger.LogInformation(JsonConvert.SerializeObject(message));
        }
    }

WindsorMessageDispatcher.cs

    /// <summary>
    /// 結合官方教程實作自動訂閱訊息轉發
    /// </summary>
    public class WindsorMessageDispatcher : IAutoSubscriberMessageDispatcher
    {
        private readonly IServiceProvider service;

        /// <summary>
        /// 引數由中間件的AutoSubscriberMessageDispatcher進行傳遞
        /// </summary>
        /// <param name="service"></param>
        public WindsorMessageDispatcher(IServiceProvider service)
        {
            this.service = service;
        }

        /// <summary>
        /// 同步方法，可以決議TConsumer對應的服務
        /// </summary>
        /// <typeparam name="TMessage"></typeparam>
        /// <typeparam name="TConsumer"></typeparam>
        /// <param name="message"></param>
        /// <param name="cancellationToken"></param>
        void IAutoSubscriberMessageDispatcher.Dispatch<TMessage, TConsumer>(TMessage message, CancellationToken cancellationToken)
        {
            //通過容器獲取對應的消費者服務，也可以通過Autofac實作
            var consumer = service.GetService<IConsume<MessageDto>>();
            //通過決議到的服務呼叫對應的消費者實作的方法
            consumer.Consume(message as MessageDto, cancellationToken);
        }

        /// <summary>
        /// 異步方法
        /// </summary>
        /// <typeparam name="TMessage"></typeparam>
        /// <typeparam name="TConsumer"></typeparam>
        /// <param name="message"></param>
        /// <param name="cancellationToken"></param>
        /// <returns></returns>
        async Task IAutoSubscriberMessageDispatcher.DispatchAsync<TMessage, TConsumer>(TMessage message, CancellationToken cancellationToken)
        {
            var consumer = service.GetService<TConsumer>();
            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<IConsume<MessageDto>,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

標籤：.NET技术

上一篇：Metalama簡介5.配合VisualStudio自定義重構或快速操作功能

下一篇：C#自定義組態檔（一）