.net core使用EasyNetQ做EventBus

男娘i 2023-10-11 10:46 28阅读 0赞

转自 https://www.cnblogs.com/focus-lei/p/9121095.html

还没有亲测，有时间反馈测试结果

随着SOA、微服务、CQRS的盛行，EventBus越来越流行，上GitHub搜了一下，还是有蛮多的这类实现，老牌的有NServiceBus（收费）、MassTransit，最近的有CAP（国人写的，1.4k个Star，非常不错）、ReBus（张队长在NanoFabric中推荐的）、RawRabbit等，今天我介绍的是另外一款产品：EasyNetQ，支持.net core，正如其名，用起来确实very easy。

首先在asp.net core项目中添加EasyNetQ的nuget引用，注入：

public void ConfigureServices(IServiceCollection services)
    {
           string rabbitMqConnection = Configuration["RabbitMqConnection"];
           services.AddSingleton(RabbitHutch.CreateBus(rabbitMqConnection));
    }

## 1、发布消息： ##

![复制代码][copycode.gif]

public class ValuesController : Controller
        {
            private IBus _bus;
            public ValuesController(IBus bus)
            {
                _bus = bus;
            }
    
    
            // POST api/values
            [HttpPost]
            public async Task Post([FromBody]Order message)
            {
                await _bus.PublishAsync(message);
            }
        }

![复制代码][copycode.gif]

可自定义消息的Exchange和Queue，需要修改消息实体：

[Queue("Qka.Order", ExchangeName = "Qka.Order")]
        public class Order
        {
            public int OrderId { get; set; }
        }

## 2、订阅消息 ##

我们订阅的服务也是寄宿在asp.net core程序中，首先定义消费者：

![复制代码][copycode.gif]

public class OrderConsumer : IConsume<Order>
        {
            [AutoSubscriberConsumer(SubscriptionId = "OrderService")]
            public void Consume(Order message)
            {
                //业务代码
            }
        }

![复制代码][copycode.gif]

也可以使用异步方法：

![复制代码][copycode.gif]

public class OrderConsumer : IConsumeAsync<Order>
        {
            [AutoSubscriberConsumer(SubscriptionId = "OrderService")]
            public Task ConsumeAsync(Order message)
            {
                //业务代码
            }
        }

![复制代码][copycode.gif]

利用EasyNetQ的自动订阅者进行订阅，这里扩展IApplicationBuilder：

![复制代码][copycode.gif]

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<IApplicationLifetime>();
                var bus = services.GetService<IBus>();
                lifeTime.ApplicationStarted.Register(() =>
                {
                    var subscriber = new AutoSubscriber(bus, subscriptionIdPrefix);
                    subscriber.Subscribe(assembly);　　　　　　　　　
                    subscriber.SubscribeAsync(assembly);
                 }); 
    
    　　　　　　　　lifeTime.ApplicationStopped.Register(() => { bus.Dispose(); }); 
    
    　　　　　　　　return appBuilder; 
    　　　　 } 
    }

![复制代码][copycode.gif]

使用扩展方法：

public void Configure(IApplicationBuilder app, IHostingEnvironment env)
            {
                app.UseSubscribe("OrderService", Assembly.GetExecutingAssembly());
            }

转载于:https://www.cnblogs.com/jzz228/p/11276893.html

[copycode.gif]: https://common.cnblogs.com/images/copycode.gif