方法3：使用一個流程管理器

圖中的數字對應於以下步驟： 1. 從UI發送一個命令用來註冊參會者X和Y到157號會議，這個命令被路由到訂單(Order)聚合。 2. 這個新的訂單(Order)聚合，ID為4239，被持久化到資料庫中 3. 訂單(Order)聚合引發(Raise)一個事件，這個事件將被RegistrationProcessManager類處理 4. RegistrationProcessManager類將發送一個命令到ID為157的可用座位(SeatsAvailability)聚合 5. 這個可用座位(SeatsAvailability)聚合從資料庫中取回還原(rehydrated) 6. 可用座位(SeatsAvailability)聚合更新自己的預定座位總數，然後持久化回資料庫

Gary(CQRS專家)發言：

流程管理器或Saga，起初，團隊將RegistrationProcessManager類看做一個Saga，但是，當他們重新閱讀Hector Garcia-Molina和Kenneth Salem合著的《Saga》一文中對「Saga」的最初定義後，他們修改了自己的決定。主要原因是預定流程並不包含明確的補償步驟，所以並不需要一個長生命周期的事務。

有關流程管理器和Saga的更多信息，請參見參考指南中的第6章「A Saga on Sagas」

團隊還明確了下列問題：

• 在哪裡驗證是否有足夠的座位可供註冊？在訂單(Order)聚合里還是可用座位(SeatsAvailability)聚合里？
• 事務邊界在哪裡?
• 當多個註冊者試圖同時下訂單時，該模型如何處理並發問題?
• 聚合根是什麼?

驗證

在登記人可以預訂座位之前，系統必須檢查是否有足夠的座位。雖然UI中的邏輯可以在發送命令之前驗證是否有足夠的可用座位，但是領域中的業務邏輯也必須執行檢查。這是因為在UI執行驗證之後到系統將命令發送到領域中的聚合時，狀態可能會發生變化。

Jana(軟體架構師)發言：

當我們在這裡談UI驗證時，我們指的是模型-視圖-控制器(MVC)執行的驗證，而不是瀏覽器前端。

在第一個模型中，驗證要麼在訂單(Order)聚合里，要麼在可用座位(SeatsAvailability)聚合里。如果是前者，則訂單(Order)聚合必須在預訂之前和引發事件之前從可用座位(SeatsAvailability)聚合中檢查當前的座位可用性。如果是後者，那麼可用座位(SeatsAvailability)聚合必須以某種方式通知訂單(Order)聚合它不能預訂座位，並且訂單(Order)聚合必須撤消(或彌補)它迄今為止完成的任何工作。

Beth(業務經理)發言：

撤銷只是現實生活中發生的許多彌補操作之一，彌補操作不僅僅局限於系統內，甚至可以是系統外的人工操作，例如：一個Contoso的職員或客戶經理打電話給註冊者們，告訴他們系統發生了一個錯誤，請他們忽略Contoso發來的最終確認郵件。

第二個模型的行為類似，除了訂單(Order)聚合和可用座位(SeatsAvailability)聚合是在會議(Conference)聚合里協作的。

在第三個模型中，使用了流程管理器，聚合通過流程管理器互相傳遞關於註冊者是否可以在當前時間進行預訂的消息。

所有這三個模型都需要實體就驗證過程進行通信，但是與流程管理器進行通信的第三個模型看起來比其他兩個模型更複雜一些。

事務邊界

在DDD中，聚合表示一致性邊界。因此，具有兩個聚合的第一個模型，級別具有兩個聚合和一個流程管理器的第三個模型將涉及兩個事務：一個在系統持久化新的訂單(Order)聚合時，另一個在系統持久化更新的可用座位(SeatsAvailability)聚合時。

備註：術語「一致性邊界」指的是你可以假設所有元素始終保持一致的邊界。

為了確保註冊者創建訂單時系統的一致性，兩個事務都必須成功。為了保證這一點，我們必須採取步驟，通過確保基礎設施可靠地向聚合傳遞消息，從而確保系統最終是一致的。

在第二個模型中，使用單一聚合，當註冊者下訂單時，我們只有一個事務。這似乎是三種模型里最簡單的一種。

並發

註冊過程發生在多用戶環境中，許多註冊者可以嘗試同時購買座位。團隊決定使用預約模式來解決註冊過程中的並發問題。在這種情況下，這意味著為註冊者最初保留了座位(然後其他註冊者無法使用這些座位)。如註冊者在超時時間內完成付款，系統保留預訂，否則，系統將取消預訂。

此預訂系統引入了對附加消息類型的需求，例如，報告註冊者已付款的事件，或報告超時發生的事件。

這個超時還要求系統在某個地方添加一個計時器來跟蹤預訂何時過期。

對這種使用消息序列和需要計時器的複雜模型，最好的辦法就是使用流程管理器。

聚合和聚合根

在訂單(Order)聚合和可用座位(SeatsAvailability)聚合這種兩個聚合里，團隊很容易識別出組成聚合的實體和聚合根。在單一聚合的模型中，選擇不是很明確：通過SeatsAvailability實體訪問Order，或者通過Order實體訪問SeatsAvailability，這似乎都不太自然。創建作為聚合根的新實體似乎沒有必要。

團隊決定採用包含流程管理器的模型，因為它提供了在這個限界上下文中處理並發需求的最佳方法。

實現細節

本節介紹訂單和註冊限界上下文中實現的一些重要特性。您或許需要獲取一份代碼的拷貝，這樣就可以跟隨我們的腳步。您可以從Download center下載它，或者在github：mspnp/cqrs-journey-code上得到它

不要期望代碼示例與參考實現中的代碼完全匹配。本章描述了CQRS過程中的一個步驟，隨著我們了解更多並重構代碼，實現可能會發生變化。

高層架構

正如我們在上一節中描述的，團隊最初決定使用CQRS模式在會議管理系統中實現預訂，但不使用事件源(Event Sourcing)。下圖顯示了實現的關鍵元素：MVC web應用程序、使用Azure SQL資料庫實例實現的數據存儲、讀寫模型和一些基礎設施組件。

備註：我們將在本節稍後的部分描述讀寫模型中發生的事情。

下面的部分與上圖中的數字相關，並提供了關於體系結構中各個元素的更多細節。

1. 使用讀模型(Read Model)查詢數據 ConferenceController類包含一個名為Display的action，該action創建一個包含特定會議信息的視圖(View)。這個控制器類使用以下代碼從讀模型里查詢：

public ActionResult Display(string conferenceCode)
{
var conference = this.GetConference(conferenceCode);

return View(conference);
}

private Conference.Web.Public.Models.Conference GetConference(string conferenceCode)
{
var repo = this.repositoryFactory();
using (repo as IDisposable)
{
var conference = repo.Query<Conference>().First(c => c.Code == conferenceCode);

var conferenceModel =
new Conference.Web.Public.Models.Conference { Code = conference.Code, Name = conference.Name, Description = conference.Description };

return conferenceModel;
}
}

讀模型(Read Model)從數據存儲中檢索信息，並使用數據傳輸對象(DTO)將信息返回給控制器。

1. 發出命令 Web應用通過命令匯流排(Command Bus)向寫模型(Write Model)發送命令。命令匯流排是系統中的可靠消息傳遞基礎設施組件。在這個場景中，它非同步將命令發送給接受者，並且只發送一次。 RegistrationController類可以向寫模型(Write Model)發送RegisterToConference命令，此命令發送一個請求，請求在會議上註冊一個或多個席位，然後，RegistrationController類輪詢讀模型(Read Model)，以發現註冊請求是否成功。參見第6節：「輪詢讀模型(Read Model)」以獲得更多細節。 下面的代碼示例展示了RegistrationController如何發送RegisterToConference命令：

var viewModel = this.UpdateViewModel(conferenceCode, contentModel);

var command =
new RegisterToConference
{
OrderId = viewModel.Id,
ConferenceId = viewModel.ConferenceId,
Seats = viewModel.Items.Select(x => new RegisterToConference.Seat { SeatTypeId = x.SeatTypeId, Quantity = x.Quantity }).ToList()
};

this.commandBus.Send(command);
備註：所有的命令都是非同步發送的，不需要等待返回。

1. 處理命令

public void Handle(RegisterToConference command)
{
var repository = this.repositoryFactory();

using (repository as IDisposable)
{
var seats = command.Seats.Select(t => new OrderItem(t.SeatTypeId, t.Quantity)).ToList();

var order = new Order(command.OrderId, Guid.NewGuid(), command.ConferenceId, seats);

repository.Save(order);
}
}

2. 在領域中初始化業務邏輯

3. 把改動持久化

4. 輪詢讀模型(Read Model)

[HttpPost]
public ActionResult StartRegistration(string conferenceCode, OrderViewModel contentModel)
{
...

this.commandBus.Send(command);

var draftOrder = this.WaitUntilUpdated(viewModel.Id);

if (draftOrder != null)
{
if (draftOrder.State == "Booked")
{
return RedirectToAction("SpecifyPaymentDetails", new { conferenceCode = conferenceCode, orderId = viewModel.Id });
}
else if (draftOrder.State == "Rejected")
{
return View("ReservationRejected", viewModel);
}
}

return View("ReservationUnknown", viewModel);
}

後來，團隊用Post-Redirect-Get模式的實現替換了這種檢查系統是否保存訂單的機制。下面的代碼示例展示了StartRegistration方法的新版本。

備註：更多關於Post-Redirect-Get模式的信息，請在Wikipedia查看Post/Redirect/Get

[HttpPost]
public ActionResult StartRegistration(string conferenceCode, OrderViewModel contentModel)
{
...

this.commandBus.Send(command);

return RedirectToAction("SpecifyRegistrantDetails", new { conferenceCode = conferenceCode, orderId = command.Id });
}

新的StartRegistration action方法現在發送命令後立即重定向到SpecifyRegistrantDetails action。下面的代碼示例顯示了SpecifyRegistrantDetails action如何在返回視圖之前輪詢資料庫中的訂單。

[HttpGet]
public ActionResult SpecifyRegistrantDetails(string conferenceCode, Guid orderId)
{
var draftOrder = this.WaitUntilUpdated(orderId);

...
}

新方法的優點：使用Post-Redirect-Get模式而不是StartRegistration post action,能讓瀏覽器的「前進」和「後退」導航按鈕工作的更好,並在控制器開始輪詢之前給命令處理程序更多時間來處理命令。

譯者註：此文章編寫時間較早。文中的UI端指的是http://asp.net mvc web應用程序。現在流行的方式是前後端分離。後端入口服務一般是一個web api程序。而且基於輪詢的方案也不太理想。可以在web api端通過消息匯流排訂閱數據持久化的各類事件，當在數據存儲層引發這些事件時。web api中的事件處理程序可以接收到視圖模型改變的數據。再將數據通過SignalR或web socket方式推送至前端。

在寫模型內部

聚合

下面是訂單(Order)聚合的代碼示例：

public class Order ： IAggregateRoot, IEventPublisher
{
public static class States
{
public const int Created = 0;
public const int Booked = 1;
public const int Rejected = 2;
public const int Confirmed = 3;
}

private List<IEvent> events = new List<IEvent>();

...

public Guid Id { get; private set; }

public Guid UserId { get; private set; }

public Guid ConferenceId { get; private set; }

public virtual ObservableCollection<TicketOrderLine> Lines { get; private set; }

public int State { get; private set; }

public IEnumerable<IEvent> Events
{
get { return this.events; }
}

public void MarkAsBooked()
{
if (this.State != States.Created)
throw new InvalidOperationException();

this.State = States.Booked;
}

public void Reject()
{
if (this.State != States.Created)
throw new InvalidOperationException();

this.State = States.Rejected;
}
}

注意類的屬性沒有全被標記為virtual。在這個類的原始版本中，屬性Id、UserId、ConferenceId和State都被標記為virtual。下面是兩個開發人員之間的討論：

• 開發人員1：我確信你不應該使屬性都成為虛擬的，除非對象關係映射(ORM)層需要。如果只是出於測試目的，實體和聚合根永遠不能用mock測試。如果你需要mock來測試實體和聚合根，那麼很明顯，設計中有問題。
• 開發人員2：在默認情況下，我更喜歡開放和可擴展性。你永遠不知道將來會出現什麼需求，把屬性標記為virtual並不費什麼事。這當然是有爭議的，在.net中有點不標準。這樣吧，我們可能只需要給延遲載入的集合屬性標記為virtual。
• 開發人員1：使用CQRS模式通常會使延遲載入的效果消失，所以你也不應該需要它。這樣會讓代碼更簡單。
• 開發人員2：CQRS並沒有說要使用事件源(Event Sourcing)，但如果使用包含對象的聚合根，無論如何都需要它，對嗎?
• 開發人員1：這不是關於Event Sourcing的，而是關於DDD的。當聚合邊界正確時，你就不需要延遲載入。
• 開發人員2：需要明確的是，聚合邊界在這裡是為了將應該一起更改的內容分組，以保持一致性。延遲載入就意味著已經分組在一起的東西其實並不需要分組。
• 開發人員1：我同意。我發現在命令端延遲載入意味著建模錯誤。如果我不需要命令端的值，那麼它就不應該在那裡。此外，我不喜歡virtual，除非它們有特定的用途(或者對象關係映射(ORM)工具的需求)。在我看來，這違反了開閉原則：你以各種可能有意也可能無意的方式敞開了自己接受修改的大門，而且即使發生了什麼影響，也可能無法立即發現。 >譯者註：ORM要求屬性必須為虛，Java里著名的Hibernate就是這麼搞得，所以NHibernate也是這樣的。
• 開發人員2：模型中的訂單聚合有一個訂單項列表。確定我們不需要載入就能把它標記為已訂好的嗎？我們建立的模型有問題嗎？
• 開發人員1：OrderItems列表很長嗎？如果是，那麼建模可能是錯誤的，因為你並不一定需要那個級別的事務。通常，較晚的來回獲取和更新OrderItems的成本可能比預先載入它們要高，你應該評估列表的通常大小，並進行一些性能度量。首先讓它變得簡單，其次如果需要的話進行優化。 -感謝Jeremie Chassaing和Craig Wilson

聚合和流程管理器

下圖展示了寫模型(Write Model)中存在的對象。有兩個聚合，Order和SeatsAvailability，每個都包含多個實體類型。此外，還有一個RegistrationProcessManager類來管理聚合之間的交互。

下圖中的表展示了流程管理器在給定當前狀態和特定類型消息時的行為。

寫模型中的領域對象

註冊會議的過程從UI發送RegisterToConference命令開始。基礎設施將此命令傳遞給訂單(Order)聚合。這個命令的結果是：系統創建了一個新的訂單(Order)聚合實例，並且這個新實例引發了一個OrderOrdered事件。訂單(Order)聚合類中的構造函數中的以下代碼示例展示了這種情況。請注意系統如何使用Guid來標識不同的實體。

public Order(Guid id, Guid userId, Guid conferenceId, IEnumerable<OrderItem> lines)
{
this.Id = id;
this.UserId = userId;
this.ConferenceId = conferenceId;
this.Lines = new ObservableCollection<OrderItem>(items);

this.events.Add(
new OrderPlaced
{
OrderId = this.Id,
ConferenceId = this.ConferenceId,
UserId = this.UserId,
Seats = this.Lines.Select(x => new OrderPlaced.Seat { SeatTypeId = x.SeatTypeId, Quantity = x.Quantity }).ToArray()
});
}

備註：要查看基礎設施組件如何傳遞命令和事件，在後面的圖裡有。

系統創建一個新的RegistrationProcessManager實例來管理新訂單。下面來自RegistrationProcessManager類的代碼示例展示了流程管理器如何處理事件。

public void Handle(OrderPlaced message)
{
if (this.State == ProcessState.NotStarted)
{
this.OrderId = message.OrderId;
this.ReservationId = Guid.NewGuid();
this.State = ProcessState.AwaitingReservationConfirmation;

this.AddCommand(
new MakeSeatReservation
{
ConferenceId = message.ConferenceId,
ReservationId = this.ReservationId,
NumberOfSeats = message.Items.Sum(x => x.Quantity)
});
}
else
{
throw new InvalidOperationException();
}
}

代碼示例顯示流程管理器如何更改其狀態，並發送一個由SeatsAvailability聚合處理的新的MakeSeatReservation命令。代碼示例還演示了如何將流程管理器實現為接收消息、更改其狀態並發送新消息的狀態機。

Markus(軟體開發人員)發言：

注意我們生成一個新的全局惟一標識符(GUID)來標識新的預訂。我們使用這些Guid將消息關聯到正確的流程管理器實例和聚合實例。

當SeatsAvailability聚合接收到MakeReservation命令時，如果有足夠的可用座位，它將進行預訂。下面的代碼示例顯示了SeatsAvailability類如何根據是否有足夠的座位引發不同的事件。

public void MakeReservation(Guid reservationId, int numberOfSeats)
{
if (numberOfSeats > this.RemainingSeats)
{
this.events.Add(new ReservationRejected { ReservationId = reservationId, ConferenceId = this.Id });
}
else
{
this.PendingReservations.Add(new Reservation(reservationId, numberOfSeats));
this.RemainingSeats -= numberOfSeats;
this.events.Add(new ReservationAccepted { ReservationId = reservationId, ConferenceId = this.Id });
}
}


流程管理器RegistrationProcessManager類處理預訂的接收和拒絕事件。這是一個臨時的座位預訂，讓用戶有機會進行支付。流程管理器在購買完成或預訂超時過期時釋放預訂。下面的代碼示例顯示流程管理器如何處理這兩種事件。

public void Handle(ReservationAccepted message)
{
if (this.State == ProcessState.AwaitingReservationConfirmation)
{
this.State = ProcessState.AwaitingPayment;

this.AddCommand(new MarkOrderAsBooked { OrderId = this.OrderId });
this.commands.Add(
new Envelope<ICommand>(new ExpireOrder { OrderId = this.OrderId, ConferenceId = message.ConferenceId })
{
Delay = TimeSpan.FromMinutes(15),
});
}
else
{
throw new InvalidOperationException();
}
}

public void Handle(ReservationRejected message)
{
if (this.State == ProcessState.AwaitingReservationConfirmation)
{
this.State = ProcessState.Completed;
this.AddCommand(new RejectOrder { OrderId = this.OrderId });
}
else
{
throw new InvalidOperationException();
}
}

如果預訂被接受，流程管理器將通過向自身發送ExpireOrder命令啟動計時器，並向訂單(Order)聚合發送MarkOrderAsBooked命令。否則，它將向訂單(Order)聚合發送一條ReservationRejected消息。

前面的代碼示例顯示了流程管理器如何發送ExpireOrder命令。基礎設施負責將消息保存在隊列中，等待15分鐘的延遲。

您可以借鑒SeatsAvailability和RegistrationProcessManager類里的代碼，以查看其他消息處理程序是如何實現的。它們都遵循相同的模式：接收消息、執行一些邏輯並發送消息。

Jana(軟體架構師)發言：

本章展示的代碼示例都來自會議管理系統的早期版本。下一章將展示當團隊持續探索該領域以及學習了更多CQRS模式的知識之後，設計和實現是如何隨之發展的。

基礎設施

下面的序列圖展示了基礎設施組件如何與領域對象交互消息的。

當UI中的MVC控制器使用命令匯流排發送消息時，典型的交互就開始了。消息發送方非同步調用命令匯流排上的Send方法。然後命令匯流排存儲消息，直到消息接收者收到消息並將其轉發給適當的處理程序。系統包含許多命令處理程序，這些命令處理程序向命令匯流排註冊，以處理特定類型的命令。例如，OrderCommandHandler類為RegisterToConference、Markorderasbooking和RejectOrder命令定義了處理程序方法。下面的代碼示例顯示了Markorderasbooking命令的處理程序方法。處理程序方法負責尋找正確的聚合實例，調用該實例上的方法，然後保存該實例。

public void Handle(MarkOrderAsBooked command)
{
var repository = this.repositoryFactory();

using (repository as IDisposable)
{
var order = repository.Find<Order>(command.OrderId);

if (order != null)
{
order.MarkAsBooked();
repository.Save(order);
}
}
}

實現IRepository介面的類負責在事件匯流排上持久化聚合對象並發布聚合里引發的任何事件，所有的這些都是事務的一部分。

Carlos(領域專家)發言：

稍後，當團隊試圖使用Azure服務匯流排作為消息傳遞基礎設施時，發現了一個問題。Azure服務匯流排不支持帶有資料庫的分散式事務。有關這個問題的討論，請參閱第5章「準備發布V1版本」。

在註冊限界上下文中，惟一的事件訂閱者是RegistrationProcessManager類。它的Router訂閱者從訂閱事件匯流排訂閱，來處理特定的事件，下面的代碼示例展示了RegistrationProcessManager類。

我們使用了術語Handler來指代處理命令和事件並將它們轉發給聚合實例的類，使用術語Router來指代處理事件和命令並將它們轉發給流程管理器實例的類。
public void Handle(ReservationAccepted @event)
{
var repo = this.repositoryFactory.Invoke();
using (repo as IDisposable)
{
lock (lockObject)
{
var process = repo.Find<RegistrationProcessManager>(@event.ReservationId);
process.Handle(@event);

repo.Save(process);
}
}
}

通常，事件處理程序方法獲取流程管理器實例，將事件傳遞給流程管理器，然後保存流程管理器實例。在本例中，IRepository實例負責持久化流程管理器實例，並負責將任何命令從流程管理器實例發送到命令匯流排。

使用Azure服務匯流排(Service Bus)

為了傳輸命令和事件，團隊決定使用Azure服務匯流排來提供底層消息傳遞基礎設施。本節描述了系統如何使用Azure服務匯流排，以及團隊在設計階段考慮的一些替代方案和權衡。

Jana(軟體架構師)發言：

Contoso的開發團隊決定使用Azure服務匯流排，因為它為會議管理系統中的消息傳遞場景提供了開箱即用的支持。這將最小化團隊需要編寫的代碼量，並提供健壯的、可伸縮的消息傳遞基礎設施。該團隊計劃使用重複消息檢測和保證消息排序等功能。要了解Azure服務匯流排和Azure隊列之間的區別，請參閱MSDN上的「Microsoft Azure Queues and Microsoft Azure Service Bus Queues - Compared and Contrasted」。

下圖顯示了命令和事件消息如何在系統中流動。MVC控制器和領域對象使用CommandBus和EventBus實例將BrokeredMessage消息發送給Azure服務匯流排中的兩個Topic之一。接收消息時，消息處理類是CommandProcessor和EventProcessor實例，CommandProcessor類確定哪個處理程序應該接收命令消息，EventProcessor類確定哪些處理程序應該接收事件消息。後者使用SubscriptionReceiver類從Topic獲取事件。處理程序實例負責調用領域對象上的方法。

Azure服務匯流排的Topic可以有多個訂閱者。Azure服務匯流排將發送到Topic的消息傳遞給它的所有訂閱者。因此，一條消息可以有多個接收者。

在最初的實現中，CommandBus和EventBus類非常相似。Send方法和Publish方法之間的惟一區別是，Send方法期望消息被包裝在Envelope類中。Envelope類允許發送方指定消息傳遞的時間延遲。

事件可以有多個接收者。在上圖的示例中，ReservationRejected事件被發送到RegistrationProcessManager、WaitListProcessManager和另一個目的地。EventProcessor類通過檢查已註冊的處理程序列表來標識收到事件的處理程序列表。

命令只有一個接收者。在上圖中，MakeSeatReservation被發送到可用座位(SeatsAvailability)聚合。只有一個為該命令註冊的處理程序。CommandProcessor類通過檢查已註冊的處理程序列表來標識收到命令的處理程序。

這一實現帶來了一些問題：

• 如何將命令的傳遞限制為單個接收?
• 如果CommandBus和EventBus類如此相似，為什麼要分別使用它們呢?
• 這種實現的可伸縮性如何?
• 這種實現的健壯性如何?
• 怎麼劃分Topic和訂閱的粒度?
• 命令和事件如何序列化?

下面幾節將討論這些問題。

將命令傳遞給單個接收者

本討論假設您已經基本了解了Azure服務匯流排隊列和Topic之間的區別。有關Azure服務匯流排的介紹，請參閱參考指南中的「參考實現中使用的技術」。

使用上圖所示的實現，有兩件事是必要的，以確保命令只有單個處理程序。首先，Azure服務匯流排中應該保證只有一個會議/命令Topic的訂閱。請記住，Azure服務匯流排主題是可以有多個訂閱者的。其次，CommandProcessor應該為它接收到的每個命令調用一個處理程序。Azure服務匯流排中沒有辦法將主題限制為單個訂閱。因此，開發人員必須自己小心的為命令的Topic創建單個訂閱。

Gary(CQRS專家)發言：

另一個問題是確保處理程序從Topic獲取命令後只處理一次。您必須確保命令是冪等的，或者系統保證只處理命令一次。該團隊將在旅程的後期處理這個問題。有關更多信息，請參見旅程第7章「增添適應能力和優化性能」。

備註：可能會運行多個SubscriptionReceiver實例，因為可以同時部署運行多個工作服務。如果多個SubscriptionReceiver實例可以接收來自同一主題訂閱的消息，那麼第一個調用SubscriptionClient對象上的Receive方法的實例將獲取並處理該命令。

另一種方法是使用Azure服務匯流排隊列代替Topic來傳遞命令。Azure服務匯流排隊列與Topic的不同之處在於，它們的設計目的是將消息傳遞給單個接收者，而不是通過多個訂閱傳遞給多個接收者。開發人員計劃更詳細的評估這個方案，以便在項目的稍後部分用此方案來實現。

下面來自SubscriptionReceiver類的代碼示例顯示了它如何接收來自Topic訂閱的消息。

private SubscriptionClient client;

...

private void ReceiveMessages(CancellationToken cancellationToken)
{
while (!cancellationToken.IsCancellationRequested)
{
BrokeredMessage message = null;

try
{
message = this.receiveRetryPolicy.ExecuteAction<BrokeredMessage>(this.DoReceiveMessage);
}
catch (Exception e)
{
Trace.TraceError("An unrecoverable error occurred while trying to receive a new message：
{0}", e);

throw;
}

try
{
if (message == null)
{
Thread.Sleep(100);
continue;
}

this.MessageReceived(this, new BrokeredMessageEventArgs(message));
}
finally
{
if (message != null)
{
message.Dispose();
}
}
}
}

protected virtual BrokeredMessage DoReceiveMessage()
{
return this.client.Receive(TimeSpan.FromSeconds(10));
}

Jana(軟體架構師)發言：

此代碼示例展示了系統如何使用Transient Fault Handling Application Block可靠地從Topic獲取消息。

Azure服務匯流排SubscriptionClient類使用peek/lock技術從訂閱中獲取消息。在代碼示例中，Receive方法在訂閱時鎖定消息。當消息被鎖定時，其他客戶端無法看到它。然後Receive方法嘗試處理消息。如果客戶端成功處理消息，則調用Complete方法：這將從訂閱中刪除消息。否則，如果客戶端未能成功處理該消息，則調用Abandon方法：這將釋放消息上的鎖，然後相同的客戶端或不同的客戶端就可以繼續接收它。如果客戶端在固定的時間內沒有調用Complete方法或Abandon方法，則也會釋放消息上的鎖。

MessageReceived事件將一個引用傳遞給SubscriptionReceiver實例，以便處理程序在處理消息時可以調用Complete方法或Abandon方法。

下面來自MessageProcessor類的代碼示例展示了如何使用BrokeredMessage實例作為MessageReceived事件的參數以及如何使用它調用Complete和Abandon方法。

private void OnMessageReceived(object sender, BrokeredMessageEventArgs args)
{
var message = args.Message;

object payload;
using (var stream = message.GetBody<Stream>())
using (var reader = new StreamReader(stream))
{
payload = this.serializer.Deserialize(reader);
}

try
{
...

ProcessMessage(payload);

...
}
catch (Exception e)
{
if (args.Message.DeliveryCount > MaxProcessingRetries)
{
Trace.TraceWarning("An error occurred while processing a new message and will be dead-lettered：
{0}", e);
message.SafeDeadLetter(e.Message, e.ToString());
}
else
{
Trace.TraceWarning("An error occurred while processing a new message and will be abandoned：
{0}", e);
message.SafeAbandon();
}

return;
}

Trace.TraceInformation("The message has been processed and will be completed.");
message.SafeComplete();
}

備註：本示例使用可靠的Transient Fault Handling Application Block，並使用擴展方法調用BrokeredMessage的Complete方法和Abandon方法。

為什麼分為CommandBus和EventBus?

儘管在會議管理系統開發的早期階段，CommandBus和EventBus類的實現非常相似，您可能想知道為什麼我們同時擁有這兩個分開的類，因為團隊預計它們在未來會出現區別。

Markus(軟體開發人員)發言：

在調用處理程序的方式和為它們捕獲什麼樣的上下文方面可能存在差異：命令可能希望捕獲額外的運行時狀態，而事件通常不需要這樣做。由於這些潛在的未來差異，我不想統一實現。我以前也遇到過這種情況，一旦有進一步的要求時，我就把它們分開。

這個方案的可擴展性如何?

使用這種方案，您可以在不同的Azure工作角色實例中運行SubscriptionReceiver類的多個實例和各種處理程序，這使您能夠擴展您的解決方案。您還可以在不同的Azure工作角色實例中擁有CommandBus、EventBus和TopicSender類的多個實例。

有關擴展Azure服務匯流排基礎設施的信息，請參閱MSDN上的 Best Practices for Performance Improvements Using Service Bus Brokered Messaging

這個方案的健壯性如何?

方案使用Azure服務匯流排的代理消息傳遞選項來提供非同步消息傳遞。服務匯流排總是可靠地存儲消息，直到用戶連接並獲取這些消息。

另外，從隊列或Topic訂閱獲取消息的peek/lock方法為消息消費者在處理消息失敗的場景中增加了可靠性。如果消費者在調用Complete方法之前失敗，則當消費者重新啟動時，任然可以處理該消息。

怎麼劃分Topic和訂閱的粒度?

當前的實現是系統中的所有命令都使用一個Topic(會議/命令)，為系統中的所有事件也使用一個Topic(會議/事件)。每個Topic都有一個訂閱，每個訂閱接收發送到該主題的所有消息。CommandProcessor和EventProcessor類負責將消息傳遞給正確的處理程序。

將來，團隊會研究使用多個Topic，例如，為每個限界上下文使用單獨的命令Topic和多個訂閱(一個事件類型一個訂閱)。這些替代方案可以簡化代碼，並促進擴展應用程序跨多個Azure工作角色，來工作。

Jana(軟體架構師)發言：

使用多個Topic、訂閱或隊列沒有額外的成本。Azure服務匯流排是根據發送的消息數量和從Azure子區域傳輸的數據量來進行計費的。

命令和事件如何序列化?

Contoso會議管理系統使用Json.NET來序列化和反序列化。有關應用程序如何使用序列化工具的詳細信息，請參閱參考指南中的「參考實現中使用的技術」

您應該考慮是否需要為命令使用Azure服務匯流排。命令通常使用在有邊界的上下文中，您可能不需要跨進程邊界發送它們(在寫入端，您可能不需要額外的層)，在這種情況下，您可以使用內存隊列來傳遞命令。」 -- Greg Young，與模式與實踐團隊的對話

對測試的影響

因為這是團隊處理的第一個限界上下文，所以關鍵一點是，如果團隊希望採用測試驅動開發(TDD)，那麼如何進行測試。下面是兩名開發人員之間的對話，他們討論了在沒有事件源(ES)的情況下實現CQRS模式時如何進行TDD，對話總結了他們的想法：

• 開發人員1：如果我們使用事件源(ES)，那麼在創建領域對象時使用TDD方法將會很容易。測試的輸入將是一個命令(可能起源於UI)，然後我們可以測試領域對象是否觸發了預期的事件。然而，如果我們不使用事件源，我們就沒有任何事件，領域對象的行為是通過ORM層將其更改持久化到數據存儲中的。
• 開發人員2：那麼我們為什麼不發起事件呢？我們沒有使用事件源(ES)並不意味著我們的領域對象不能引發事件。讓領域對象引發事件，然後我們可以按照通常的方法設計測試，以檢查響應命令時觸發的正確事件。
• 開發人員1：這難道不是讓事情變得比需要的更複雜了嗎？使用CQRS的動機之一就是簡化事情！現在我們有了領域對象，它們需要使用ORM層來持久化它們的狀態。然後我們又要引發事件來報告它們所持久化的內容，因為這樣我們就可以運行單元測試了。
• 開發人員2：我明白你的意思。
• 開發人員1：我們可能在如何進行測試上遇到了瓶頸。也許我們不應該基於領域對象的預期行為來設計測試，而是應該考慮在領域對象處理命令之後測試它們的狀態。
• 開發人員2：這應該很容易做到，畢竟，領域對象把我們想要檢查的所有數據都存儲在屬性中，以便ORM可以將正確的信息持久化到存儲中。
• 開發人員1：所以我們只需要考慮在這個場景中使用另一種不同的風格進行測試。
• 開發人員2：我們還需要考慮這個問題的另一個方面：我們可能有一組測試來測試領域對象，並且所有這些測試都可能通過。我們還可能有一組測試來驗證ORM層是否能夠成功地保存和獲取對象。但是，我們還必須測試領域對象在ORM層上運行時是否正確。領域對象有可能執行正確的業務邏輯，但無法正確的持久化其狀態，這可能是因為ORM處理特定數據類型的方式存在問題。

有關這裡討論的兩種測試方法的更多信息，請參閱Martin Fowler的文章「Mocks Arent Stubs」和Steve Freeman、Nat Pryce和Joshua Kerievsky編寫的「Point/Counterpoint」。

備註：解決方案中包含的測試是使用xUnit.net編寫的。

下面的代碼示例展示了使用上面討論的行為方法編寫的兩個測試示例。

Markus(軟體開發人員)發言：

這些是我們剛開始時使用的測試，但是我們隨後用基於狀態的測試替換了它們。

public SeatsAvailability given_available_seats()
{
var sut = new SeatsAvailability(SeatTypeId);
sut.AddSeats(10);
return sut;
}

[TestMethod]
public void when_reserving_less_seats_than_total_then_succeeds()
{
var sut = this.given_available_seats();
sut.MakeReservation(Guid.NewGuid(), 4);
}

[TestMethod]
[ExpectedException(typeof(ArgumentOutOfRangeException))]
public void when_reserving_more_seats_than_total_then_fails()
{
var sut = this.given_available_seats();
sut.MakeReservation(Guid.NewGuid(), 11);
}


這兩個測試共同驗證了可用座位(SeatsAvailability)聚合的行為。在第一個測試中，預期的行為是MakeReservation方法成功，並且不會拋出異常。在第二個測試中，MakeReservation方法的預期行為是拋出異常，因為沒有足夠的空閑座位來完成預訂。

如果沒有聚合引發事件，則很難以任何其他方式測試行為。例如，檢查是否進行了正確的調用以將聚合持久化到數據存儲里，如果您試圖用這個來測試行為，那麼測試就會和數據存儲實現耦合(這是一種壞氣味)：如果希望更改數據存儲的實現，那麼就需要更改領域模型中對聚合的測試。

下面的代碼示例展示了使用被測試對象的狀態編寫的測試示例。這是在項目中使用的一種測試風格。

public class given_available_seats
{
private static readonly Guid SeatTypeId = Guid.NewGuid();

private SeatsAvailability sut;
private IPersistenceProvider sutProvider;

protected given_available_seats(IPersistenceProvider sutProvider)
{
this.sutProvider = sutProvider;
this.sut = new SeatsAvailability(SeatTypeId);
this.sut.AddSeats(10);

this.sut = this.sutProvider.PersistReload(this.sut);
}

public given_available_seats()
: this(new NoPersistenceProvider())
{
}

[Fact]
public void when_reserving_less_seats_than_total_then_seats_become_unavailable()
{
this.sut.MakeReservation(Guid.NewGuid(), 4);
this.sut = this.sutProvider.PersistReload(this.sut);

Assert.Equal(6, this.sut.RemainingSeats);
}

[Fact]
public void when_reserving_more_seats_than_total_then_rejects()
{
var id = Guid.NewGuid();
sut.MakeReservation(id, 11);

Assert.Equal(1, sut.Events.Count());
Assert.Equal(id, ((ReservationRejected)sut.Events.Single()).ReservationId);
}
}

這裡展示的兩個測試在調用MakeReservation方法後測試可用座位(SeatsAvailability)聚合的狀態。第一個用來測試有足夠座位可用的場景。第二個用來測試沒有足夠的座位可用的場景。第二個測試可以利用可用座位(SeatsAvailability)聚合的行為，因為如果該聚合拒絕預訂，它確實會引發一個事件。

匯總

在旅程的第一階段，我們探索了實現CQRS模式的一些基礎知識，並為下一階段做了一些準備。

下一章將描述我們如何擴展和增強已經完成的工作，為訂單和註冊限界上下文添加更多的特性和功能。我們還將研究一些額外的測試技術，以了解它們可能如何幫助我們實現這一目標。

推薦閱讀：