NServiceBus 长时间运行的进程超时答案

【问题标题】：NServiceBus long running process that timesoutNServiceBus 长时间运行的进程超时
【发布时间】：2013-08-09 17:09:50
【问题描述】：

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

这是我们重现问题的方式：

将本地机器配置更新为事务的默认超时 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还在睡觉。工作线程 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
    }

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

【问题讨论】：

标签： nservicebus nservicebus3

【解决方案1】：

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

关于另一个线程问题：当线程 A 处于睡眠状态时，线程 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'睡眠继续使用相同的消息'm1'并在它尝试回复时抛出。当线程'A'处于休眠状态时，我没有看到线程'B'接听'm1'。

干杯。

【讨论】：

我们正在运行 NSB 3.3.8，我们在生产环境中遇到了这个问题，我在 3.3.8 示例中隔离并复制了这个问题。我不确定为什么您没有看到我们看到的相同结果？唯一的区别是我只发送 1 条消息？

【解决方案2】：

事务超时有类似的问题，但不想更改每个事务的默认超时，所以最终只是更改了 NServiceBus 的事务配置。

Configure.Transactions.Advanced(x => x.DefaultTimeout(TimeSpan.FromMinutes(3)));

这并不理想，建议将其分解为更多模块化的工作。

【讨论】：