nservicebus and eventstore

Question

我想知道是否有人遇到过这种情况:

我处理一个命令，在处理程序中，我将一个事件保存到eventstore (joliver)。
在调度之后，将再次处理同一命令的处理程序。
我知道这是同一个命令，因为命令上的guid是相同的。

尝试五次后，nservicebus说命令失败，因为重试次数最多。
所以很明显命令失败了，但是我没有得到任何失败的指示。我将调度程序的内容放入try catch中，但是没有捕获错误。在代码退出调度程序后，事件处理程序将始终触发，就好像出错了一样。

通过代码跟踪，事件被保存到数据库(我看到行)，调度程序运行，分派列设置为true，然后处理程序再次处理命令，重复该过程，并将另一行插入到提交表中。

到底是什么出了问题?我没有在事件存储中设置成功标志吗?如果我将eventstore与nServicebus解耦，两者都将按预期运行，没有重试和失败。

分配器:

    public void Dispatch(Commit commit)
    {
        for (var i = 0; i < commit.Events.Count; i++)
        {
            try
            {
                var eventMessage = commit.Events[i];
                var busMessage = (T)eventMessage.Body;
                //bus.Publish(busMessage);
            }
            catch (Exception ex)
            {
                throw ex;
            }
        }
    }

Wireup.Init ()

    private static IStoreEvents WireupEventStore()
    {
        return Wireup.Init()
            .LogToOutputWindow()
            .UsingSqlPersistence("EventStore")
            .InitializeStorageEngine()
            .UsingBinarySerialization()
            //.UsingJsonSerialization()
            //    .Compress()
            //.UsingAsynchronousDispatchScheduler()
            //    .DispatchTo(new NServiceBusCommitDispatcher<T>())
           .UsingSynchronousDispatchScheduler()
               .DispatchTo(new DelegateMessageDispatcher(DispatchCommit))
            .Build();
    }

Answer 1

我在保存中打开了一个事务范围，但从未关闭。

    public static void Save(AggregateRoot root)
    {
        // we can call CreateStream(StreamId) if we know there isn't going to be any data.
        // or we can call OpenStream(StreamId, 0, int.MaxValue) to read all commits,
        // if no commits exist then it creates a new stream for us.
        using (var scope = new TransactionScope())
        using (var eventStore = WireupEventStore())
        using (var stream = eventStore.OpenStream(root.Id, 0, int.MaxValue))
        {
            var events = root.GetUncommittedChanges();
            foreach (var e in events)
            {
                stream.Add(new EventMessage { Body = e });
            }
            var guid = Guid.NewGuid();
            stream.CommitChanges(guid);
            root.MarkChangesAsCommitted();
            scope.Complete(); // <-- missing this
        }
    }