NServiceBus 长时间运行的进程超时

Question

我们注意到具有长时间运行请求的消息处理程序存在一个问题，如果在分布式事务超时之前请求没有完成处理，则在原始消息处理完成之前或原始消息之前在新线程上重新处理相同的消息引发错误。该消息最终会进入错误队列。

以下是我们重现该问题的方式:

将本地机器配置更新为交易的默认超时时间为 1 分钟:

<system.transactions>        
   <defaultSettings distributedTransactionManagerName="" timeout="00:01:00" />       
    <machineSettings maxTimeout="00:01:00" />
</system.transactions>

将 NSB 消息服务配置为多线程:

<MsmqTransportConfig NumberOfWorkerThreads="2" MaxRetries="2" />

然后在我们的处理程序中让线程休眠 2 分钟(下面的代码)，发生的情况是一条消息在工作线程 A 上传入。一旦发生 1 分钟超时，工作线程 B 上处理相同的消息，工作线程 A 仍然是 sleep 。工作线程 A 最终返回并出现“无法登记事务”的错误。然后再次重新开始处理消息，这会在两个工作线程上继续进行，直到消息最终进入错误队列。

public void Handle(RequestDataMessage message)
    {
        Logger.Info("==========================================================================");
        Logger.InfoFormat("Received request {0}.", message.DataId);
        Logger.InfoFormat("String received: {0}.", message.String);
        Logger.InfoFormat("Header 'Test' = {0}.", message.GetHeader("Test"));
        Logger.InfoFormat(Thread.CurrentPrincipal != null ? Thread.CurrentPrincipal.Identity.Name : string.Empty);

        var response = Bus.CreateInstance<DataResponseMessage>(m => 
        { 
            m.DataId = message.DataId;
            m.String = message.String;
        });

        response.CopyHeaderFromRequest("Test");
        response.SetHeader("1", "1");
        response.SetHeader("2", "2");

        Thread.Sleep(new TimeSpan(0, 2, 0));



        Logger.Info("========== Thread continued ===========");
        Logger.InfoFormat("Received request {0}.", message.DataId);
        Logger.InfoFormat("String received: {0}.", message.String);
        Logger.InfoFormat("Header 'Test' = {0}.", message.GetHeader("Test"));
        Logger.InfoFormat(Thread.CurrentPrincipal != null ? Thread.CurrentPrincipal.Identity.Name : string.Empty);

        Bus.Reply(response); //Try experimenting with sending multiple responses
    }

这是一个已知问题，还是我们应该设计我们的代码以避免发生这种类型的密集再处理？

Answer 1

一般而言，将交易搁置这么长时间并不是一个好主意。这意味着参与该事务的所有资源都被锁定。您应该尝试拆分处理程序中正在进行的工作。
处理程序处理什么类型的工作？是否可以拆分该工作？

关于另一个线程问题:当线程 A 处于 sleep 状态时，线程 B 正在处理相同的消息，我在 3.3.8 版或 4.0.2 版中都没有看到这种行为

这是我的处理程序:

    public void Handle(ProcessSomething message)
    {
        Console.WriteLine("{0} -- Thread {1} Handling MessageId {2}", DateTime.Now, Thread.CurrentThread.ManagedThreadId, Bus.CurrentMessageContext.Id);

        var response = Bus.CreateInstance<DataResponseMessage>(m =>
        {
            m.DataId = message.DataId;
            m.String = message.String;
        });

        Thread.Sleep(new TimeSpan(0, 1, 10));
        Console.WriteLine("{0} -- Thread {1} contiuned after sleep MessageId {2}", DateTime.Now, Thread.CurrentThread.ManagedThreadId, Bus.CurrentMessageContext.Id);

        Bus.Reply(response); //Try experimenting with sending multiple responses
    }

我在队列中放置了 4 条消息，将线程数/MaxConcurrency 设置为 2，并将 machine.config maxtimeout 设置为 1 分钟，并将处理程序设置为休眠 1 分 10 秒。

Bus.Reply 将无法在事务中登记。因为事务超时。这会导致异常并重试消息。在同一个处理程序中发送的所有消息都是同一个事务的一部分。因此，此消息失败并重试。
我看到的是线程 'A' 处理 Msmq 消息 Id 'm1' sleeps 继续使用相同的消息 'm1' 并在它尝试回复时抛出。当线程 'A' 正在休眠时，我没有看到线程 'B' 拾取了 'm1'。

干杯。