带参数的 MassTransit 活动故障答案

【问题标题】：MassTransit Activity Fault with parameters带参数的 MassTransit 活动故障
【发布时间】：2018-01-23 18:15:27
【问题描述】：

我目前正在使用 Courier 模式的 Masstransit。

我已经设置了一个可能失败的活动，我希望能够订阅这个失败并采取相应的行动。

我的问题是，即使我可以订阅失败，甚至看到导致失败的异常，我也无法向它传递任何参数。

出于测试目的，假设我有以下活动：

public class MyActivity : ExecuteActivity<MyMessage>
{
    public Task<ExecutionResult> Execute(ExecuteContext<MyMessage> context)
    {
        try
        {
           // .... some code
            throw new FaultException<RegistrationRefusedData>(
                new RegistrationRefusedData(RegistrationRefusedReason.ItemUnavailable));
            // .... some code
        }
        catch (Exception ex)
        {
            return Task.FromResult(context.Faulted(ex));
        }
    }
}

问题在于我作为异常参数传递的原因 (RegistrationRefusedReason)。如果我订阅 RoutingSlipActivityFaulted 消费者，我可以几乎获得我需要的所有信息：

public class ActivityFaultedConsumer : IMessageConsumer<RoutingSlipActivityFaulted>
{
    public void Consume(RoutingSlipActivityFaulted message)
    {
        string exceptionMessage = message.ExceptionInfo.Message; // OK
        string messageType = message.ExceptionInfo.ExceptionType; // OK
        RegistrationRefusedReason reason =  ??????;
    }
}

我觉得我在这里遗漏了一些重要的东西，（也许滥用了模式？）。

还有其他方法可以从错误的活动中获取参数吗？

【问题讨论】：

我想如果我在活动中有一个context.FaultedWithVariables 会起作用...

标签： c# rabbitmq message-queue masstransit

【解决方案1】：

所以，您所描述的情况不是Fault。这是一个未能满足业务条件。在这种情况下，您不想重试事务，而是想终止它。要通知发送单的发起人，您需要Publish 一个业务事件，表明由于业务状况导致交易未完成。

例如，在您的情况下，您可以执行以下操作：

context.Publish<RegistrationRefused>(new {
    CustomerId = xxx,
    ItemId = xxxx,
    Reason = "Item was unavailable"
    });

context.Terminate();

这将终止路由单（不会执行后续活动），并产生RoutingSlipTerminated 事件。

这是结束因业务条件或规则而产生的传送单的正确方法。异常仅针对异常行为，因为您可能希望重试它们以处理失败。

【讨论】：

我想这是有道理的。但是，Terminate 方法会开始补偿以前的活动吗？我需要整个路由单才能进行一致的交易...
刚刚测试过......它没有。在这种情况下，我应该发送消息并使用 Faulted()，还是我在滥用它？（顺便说一句，在 mt 上做得很好）
是的，如果您想补偿，请抛出异常并仍然发布您的事件以具有业务上下文。

【解决方案2】：

有点起死回生，但我真的没有找到解决这个问题的好办法。

这是我的场景：

我想实现请求/响应，但我想等待路由单的执行。
作为 Fabio，我想补偿之前的任何活动，并且我想在出现故障时将数据传回请求客户端。

方便的是，Chris 提供了一个 RoutingSlipRequestProxy/RoutingSlipResponseProxy，它就是这样做的。我找到了 2 种方法，但对我来说，这两种方法都非常老套。

方法一：

请求客户端等待ISimpleResponse或ISimpleFailResponse。
RoutingSlipRequestProxy 在变量中设置ResponseAddress。
活动将ISimpleFailResponse 发送到ResponseAddress。
客户端等待任一响应
RoutingSlipResponseProxy 将Fault<ISimpleResponse> 发送回ResponseAddress。

据我所知，hackiness 来自步骤 4/5 及其顺序。我很确定它可以工作，但如果消息被乱序消费，它很容易停止工作。

示例代码：https://github.com/steliyan/Sample-RequestResponse/commit/3fcb196804d9db48617a49c7a8f8c276b47b03ef

方法2：

请求客户端等待ISimpleResponse或ISimpleFailResponse。
活动使用变量调用ReviseItirery 并添加错误活动。*
错误的活动故障
RoutingSlipResponseProxy2 获取ValidationErrors 并将ISimpleFailResponse 发送回ResponseAddress。

* 活动必须是Activity 而不是ExecuteActivity，因为ReviseItinerary 没有带变量但没有活动日志的重载。

这种方法看起来很老套，因为在行程中添加了一个额外的故障活动，只是为了能够将变量添加到路由单。

示例代码：https://github.com/steliyan/Sample-RequestResponse/commit/e9644fa683255f2bda8ae33d8add742f6ffe3817

结论： 查看 MassTransit 代码，添加 FaultedWithVariables 重载似乎不是问题。不过，我认为 Chris 的观点是应该有更好的方法来设计工作流程，但我不确定。

【讨论】：