nservicebus - 是什么导致 EventStore 如此容易抛出 ConcurrencyException？

Question

使用 JOliver EventStore 3.0，并且刚刚开始使用简单的示例。

我有一个使用 NServiceBus 的简单发布/订阅 CQRS 实现。客户端在总线上发送命令，域服务器接收并处理命令并将事件存储到事件存储，然后由事件存储的调度程序在总线上发布。然后读取模型服务器订阅这些事件以更新读取模型。没什么特别的，几乎是书本上的。

它正在工作，但只是在简单的测试中，当事件存储到 EventStore 时，我在域服务器上收到了很多并发异常(间歇性)。它会正确重试，但有时会达到 5 次重试限制，并且命令最终会出现在错误队列中。

我可以从哪里开始调查以查看导致并发异常的原因？我删除了调度程序，只专注于存储事件，它也有同样的问题。

我正在使用 RavenDB 来持久化我的 EventStore。我没有做任何花哨的事情，只是这个:

using (var stream = eventStore.OpenStream(entityId, 0, int.MaxValue))
{
  stream.Add(new EventMessage { Body = myEvent });
  stream.CommitChanges(Guid.NewGuid());
}

异常的堆栈跟踪如下所示:

2012-03-17 18:34:01,166 [Worker.14] WARN NServiceBus.Unicast.UnicastBus [(null)] <(null)> - EmployeeCommandHandler failed handling message. EventStore.ConcurrencyException: Exception of type 'EventStore.ConcurrencyException' was thrown. at EventStore.OptimisticPipelineHook.PreCommit(Commit attempt) in c:\Code\public\EventStore\src\proj\EventStore.Core\OptimisticPipelineHook.cs:line 55 at EventStore.OptimisticEventStore.Commit(Commit attempt) in c:\Code\public\EventStore\src\proj\EventStore.Core\OptimisticEventStore.cs:line 90 at EventStore.OptimisticEventStream.PersistChanges(Guid commitId) in c:\Code\public\EventStore\src\proj\EventStore.Core\OptimisticEventStream.cs:line 168 at EventStore.OptimisticEventStream.CommitChanges(Guid commitId) in c:\Code\public\EventStore\src\proj\EventStore.Core\OptimisticEventStream.cs:line 149 at CQRSTest3.Domain.Extensions.StoreEvent(IStoreEvents eventStore, Guid entityId, Object evt) in C:\dev\test\CQRSTest3\CQRSTest3.Domain\Extensions.cs:line 13 at CQRSTest3.Domain.ComandHandlers.EmployeeCommandHandler.Handle(ChangeEmployeeSalary message) in C:\dev\test\CQRSTest3\CQRSTest3.Domain\ComandHandlers\Emplo yeeCommandHandler.cs:line 55

Answer 1

我想到了。不得不挖掘源代码才能找到它。我希望这有更好的记录!这是我的新 eventstore 接线:

EventStore = Wireup.Init()
          .UsingRavenPersistence("RavenDB")
          .ConsistentQueries()
          .InitializeStorageEngine()
          .Build();

我必须添加 .ConsistentQueries() 以便 raven 持久性提供程序在内部使用 WaitForNonStaleResults 对事件存储对 raven 进行的查询。

基本上，当我添加一个新事件，然后在 raven catch 索引之前尝试添加另一个事件时，流修订版不是最新的。第二个事件将踩在第一个事件上。