大众运输消费者死亡-Peek()

Question

我对MassTransit使用者是否可以在实际检索msg之前可以Peek()一个MSMQ队列感到好奇。

步骤/过程是什么:

1)消息已发送到队列

2)消费者获取它并必须进行数据库更新-大约需要5秒钟

3)如果第一轮工作，消费者必须进行第二轮更新。

我的问题是，如果第一次数据库更新失败，则消息会停留在队列中(即网络问题，无法到达数据库)，该如何处理。

当前，一旦它从队列中读取了消息，它就会将其删除，然后，如果数据库更新失败，它就会消失。

此外，我该如何处理电源故障-我的意思是，如果消费者在完成“工作”的途中途(无论是什么(数据库更新或其他)以及电源中断等)，如何重新运行该流程？队列中的消息？可以说，这项工作(无论如何，在我目前的情况下)是将新行推送到表中。我的意思是我可以编写代码来首先检查该行是否存在，然后是否删除该消息，如果不是，则运行该任务，但是我如何才能首先获取它以重新运行整个过程？

我已经读过我可以Peek()队列，然后运行任务，然后读取队列中的msg并将其删除，但是我一生都无法弄清楚这是否适用于大众运输...有点丢失。 。

另外，我知道Masstransit有.RetryLater，但是我是否在此过程中使用了它？是传奇中的Initially-> When-> Then-> .RetryLater吗？

任何指针将不胜感激

最亲切的问候
罗宾

编辑

PS:我正在使用一个传奇。

Define(() =>
            {
                RemoveWhen(saga => saga.CurrentState == Completed);

                Initially(
                    When(NewAC)
                        .Then((saga, message) => saga.ProcessPSM(message),
                            InCaseOf<Exception>()
                               .TransitionTo(Problem)                                  
                                )
                        .Then((saga, message) => saga.PostProcessPSM())
                        .Complete()
                    );
                During(Problem,
                    When(Waiting)
                               // NOTE: THIS DOES NOT WORK!!!!
                        .RetryLater()
                    );
                });

RetryLater引发以下错误:
“现有的传奇无法接受该消息”

我不确定我还能如何访问“RetryLater”。

Answer 1

MassTransit抽象了基础队列的概念。因此，Peek不是解决方案，
但是它还有其他重试消息的方法。如果您只对处理错误和故障情况感兴趣，则可以使用以下机制。

默认情况下，如果使用者抛出异常，则该消息将重试N次:

，其中N在总线上配置，默认为5。可以在以下位置更改
使用ServiceBusConfigurator

上的SetDefaultRetryLimit初始化总线

重试的位置意味着邮件将被添加到队列的末尾

如果您希望使用更细粒度的错误处理方法，则可以实现Context Consumer，捕获可恢复或临时异常，并手动调用RetryLater。据我了解，可以完成多少次没有限制。

public class RetryConsumer : Consumes<AwesomeMessage>.Context
{

    public void Consume(IConsumeContext<AwesomeMessage> message)
    {
        try
        {
            Console.WriteLine("This is Attempt " + message.RetryCount);
            // Do Something
        }
        catch (SomeTransientException e)
        {
            message.RetryLater();
        }
    }
}