重新整理 .net core 實踐篇—————微服務的橋樑EventBus[三十一]

阿新 • • 發佈：2021-06-27

前言

簡單介紹一下EventBus.

正文

EventBus 也就是整合事件，用於服務與服務之間的通訊。

比如說我們的訂單處理事件，當訂單處理完畢後，我們如果通過api馬上去呼叫後續介面。

比如說訂單完成給使用者通知的話，如果是大量訂單，即使我們使用非同步async await 這種模式，在這個訂單服務中將會大量佔用資源，因為async await 本身是執行緒池。

因為裡面的資源是有限的，如果建立訂單還有完成訂單佔用大量資源的話，傳送郵件還加入到競爭中，那麼可以想象，機器掉入資源用盡就更大了。如果是訂單完成之後要呼叫多個服務，那麼可想而知，壓力多大。

那麼EventBus 就是通過釋出訂閱這種模式來快取起來，EventBus 釋出一個事件，然後其他微服務可以進行訂閱，這樣就緩解了機器的負擔。

.net core 實現EventBus 的框架如下:https://github.com/dotnetcore/CAP

可以去看下上面這個地址，這裡就不詳細介紹了，因為沒有什麼比文件更加詳細了。

在.net core 中加入EventBus:

// 註冊 EventBus  
services.AddEventBus(Configuration);

AddEventBus 如下:

/// <summary>
/// 註冊EventBus（整合事件處理服務）
/// </summary>
/// <param name="services"></param>
/// <param name="configuration"></param>
/// <returns></returns>
public static IServiceCollection AddEventBus(this IServiceCollection services, IConfiguration configuration)
{
	// 注入整合事件訂閱服務
	services.AddTransient<ISubscriberService, SubscriberService>();

	// 注入CAP服務
	services.AddCap(options =>
	{
		// 指定CAP元件所使用的資料庫上下文，當前設定表示EventBus與領域驅動共享資料庫連結
		options.UseEntityFramework<DomainContext>();

		// package: DotNetCore.CAP.RabbitMQ
		// 指定RabbitMQ作為我們EventBus的訊息佇列的儲存，並注入配置
		options.UseRabbitMQ(options =>
		{
			configuration.GetSection("RabbitMQ").Bind(options);
		});
		// options.UseDashboard();
	});
	return services;
}

可以看到上面使用RabbitMQ作為Eventbus的訊息佇列處理。當然除了RabbitMq，這個框架還支援其他的，如Kafka, AzureService, AmazonSQS等，看公司或者專案需要什麼就行。

因為使用了RabbitMq,那麼上面的寫了載入RabbitMq:configuration.GetSection("RabbitMQ").Bind(options),配置RabbitMQ:

"RabbitMQ": {
"HostName": "127.0.0.1",
"UserName": "mq",
"Password": "123456",
"VirtualHost": "blog",
"ExchangeName": "blog_queue"
}

那麼看一下EventBus釋出:

/// <summary>
/// 建立Order領域事件處理
/// </summary>
public class OrderCreatedDomainEventHandler : IDomainEventHandler<OrderCreatedDomainEvent>
{
	ICapPublisher _capPublisher;
	public OrderCreatedDomainEventHandler(ICapPublisher capPublisher)
	{
		_capPublisher = capPublisher;
	}

	/// <summary>
	/// 領域事件處理
	/// </summary>
	/// <param name="notification"></param>
	/// <param name="cancellationToken"></param>
	/// <returns></returns>
	public async Task Handle(OrderCreatedDomainEvent notification, CancellationToken cancellationToken)
	{
		// 當建立新訂單時，向 EventBus 釋出一個事件
		await _capPublisher.PublishAsync("OrderCreated", new OrderCreatedIntegrationEvent(notification.Order.Id));
	}
}

前面訂單建立完畢的時候，經過呼叫相關的領域事件，然後向EventBus 傳送了一個事件。那麼先看下這個事件。

/// <summary>
/// 整合事件：訂單建立完成
/// </summary>
public class OrderCreatedIntegrationEvent
{
	public long OrderId { get; }
	public OrderCreatedIntegrationEvent(long orderId)
	{
		OrderId = orderId;
	}
}

這樣就建立了一個整合事件，並且釋出出去了。

然後建立訂閱服務:

/// <summary>
/// 整合事件訂閱服務
/// </summary>
public class SubscriberService : ISubscriberService, ICapSubscribe
{
	IMediator _mediator;
	public SubscriberService(IMediator mediator)
	{
		_mediator = mediator;
	}

	/// <summary>
	/// 訂閱訂單建立成功整合事件
	/// </summary>
	/// <param name="event"></param>
	[CapSubscribe("OrderCreated")]
	public void OrderCreatedSucceeded(OrderCreatedIntegrationEvent @event)
	{
		// do something...
	}

	/// <summary>
	/// 訂閱訂單支付成功整合事件
	/// </summary>
	/// <param name="event"></param>
	[CapSubscribe("OrderPaymentSucceeded")]
	public void OrderPaymentSucceeded(OrderPaymentSucceededIntegrationEvent @event)
	{
		// do something...
	}
}

這樣即可，框架會幫我們處理。