【发布时间】：2022-02-09 18:21:36
【问题描述】：
这与延迟消息传递无关。最近我注意到，在查看从 RabbitMQ 队列中获取消息的时间时，我的端点在处理消息时存在延迟。
我想知道是否有一些我忽略的配置选项会影响 NServiceBus 从 RabbitMQ 接收消息与在 IHandleMessages 处理程序中实际处理消息（命令）之间的延迟。
在比较 NServiceBus 日志文件和处理程序的日志文件时，我看到了一致的延迟。它在客户端和服务器之间确实有所不同，但在给定的会话中它总是不变的。现在我正在浏览所有我能找到的选项，但在野外没有真正的指针。
问题是我正在运行一个工作流解决方案，它使用 NServiceBus 在工作流服务器（基于 MS WF）和 50/60 客户端之间进行所有消息传递。服务器在 NServiceBus 7.6.0 上运行，但所有客户端仍使用 NServiceBus 5.2.26。我决定编写一个小型测试客户端，它也可以在 NServiceBus 7.6.0 上运行，但这也显示了延迟，尽管延迟较小。但是在测试客户端中，单个命令之间的延迟是恒定的 6.5 秒。因此消息由 NSB 从队列中检索，并在 6.5 秒后调用 IHandleMessages 处理程序。
我尝试了几件事（直接路由与传统路由，例如确保许可证文件正确分发）。我现在唯一能想到的就是为我正在使用的工作流服务器创建一个新的测试服务器，它只能接受 1 个命令，生成回复，仅此而已，看看这是否会有所作为.
在配置方面，该解决方案在具有 4 个内核、RabbitMQ 3.7.21 和 NServiceBus 7.6.0 的 Windows Server 2016 上运行。客户端在使用 NServiceBus 5.2.26 的 Windows 10 机器上运行，但测试客户端使用 NServiceBus 7.6.0。
现在是日志。这是 NServicebus 收到 GetVersionCommand 时的日志：
2022-01-20 16:09:58.104 DEBUG Processing message type: HyFlo.Service.Messages.Commands.Info.GetVersionCommand
Message headers:
CorrelationId : 619d678d-25bf-46c0-b5c0-57c624e3d557
NServiceBus.MessageId : 548813d5-2465-4ff5-904d-ae2300f9ee23
NServiceBus.MessageIntent : Send
NServiceBus.ConversationId : 367575b0-7bee-40d0-90ca-ae2300f9ee24
NServiceBus.CorrelationId : 548813d5-2465-4ff5-904d-ae2300f9ee23
NServiceBus.TimeToBeReceived : 00:10:00
NServiceBus.ReplyToAddress : Hyflo.Client.Test
NServiceBus.OriginatingMachine : DYP297
NServiceBus.OriginatingEndpoint : Hyflo.Client.Test
$.diagnostics.originating.hostid : 3a3bbd6661214b655aa6dbcd95d112a2
NServiceBus.ContentType : text/xml
NServiceBus.EnclosedMessageTypes : HyFlo.Service.Messages.Commands.Info.GetVersionCommand, HyFlo.Service.Messages, Version=4.10.3.26734, Culture=neutral, PublicKeyToken=null;HyFlo.Types.Interfaces.IInfoCommand, HyFlo.Types, Version=4.10.8.20782, Culture=neutral, PublicKeyToken=null
NServiceBus.Version : 7.6.0
NServiceBus.TimeSent : 2022-01-20 15:09:58:107461 Z
NServiceBus.NonDurableMessage : False
Handlers to invoke:
HyFlo.Service.CommandHandlers.Info.GetVersionHandler
然后是GetVersionHandler的自定义日志：
2022-01-20 16:10:04.6516 TRACE = New Message =====================================================================================================================================
2022-01-20 16:10:04.6516 TRACE CommandEventId:9125b036-c315-484e-9948-889ed1e56587: New message handled by handler of type 'HyFlo.Service.CommandHandlers.Info.GetVersionHandler' with CorrelationId '619d678d-25bf-46c0-b5c0-57c624e3d557' ..
2022-01-20 16:10:04.6516 TRACE ===================================================================================================================================================
2022-01-20 16:10:04.6516 INFO Cache contains a systemstatus ..
2022-01-20 16:10:04.6516 INFO Retrieved systemstatus from cache is: 0
2022-01-20 16:10:04.6516 INFO Running the injected code ..
2022-01-20 16:10:04.6672 TRACE ===================================================================================================================================================
2022-01-20 16:10:04.6672 TRACE CommandEventId:9125b036-c315-484e-9948-889ed1e56587: Message processed by handler of type 'HyFlo.Service.CommandHandlers.Info.GetVersionHandler' with CorrelationId '619d678d-25bf-46c0-b5c0-57c624e3d557' ..
2022-01-20 16:10:04.6672 TRACE Elapsed time: 00:00:00.0156159 ..
2022-01-20 16:10:04.6672 TRACE = End of Message ==================================================================================================================================
该命令在 2022-01-20 16:09:58.104 收到，但在 2022-01-20 16:10:04.6516 在 Handler 处交付，此延迟在此命令和其他命令之间非常一致。我只是无法理解为什么会出现这种延迟。
端点配置如下：
internal class ProgramService : ServiceBase
{
private IEndpointInstance _endpointInstance = null;
/// <summary>
/// Determines if application is ran as service or just as console application
/// </summary>
/// <param name="name"></param>
/// <returns>true if it's a service</returns>
private static bool IsService(string name)
{
if (!Environment.UserInteractive) return true;
var sc = new ServiceController(name);
try
{
return sc.Status == ServiceControllerStatus.StartPending;
}
catch (InvalidOperationException)
{
return false;
}
}
/// <summary>
/// Main entry
/// </summary>
private static void Main()
{
using (var service = new ProgramService())
{
// so we can run interactive from Visual Studio or as a windows service
if (!IsService("HyFlo.Service"))
{
Console.CancelKeyPress += (sender, e) => { service.OnStop(); };
service.OnStart(null);
Console.WriteLine("\r\nPress enter key to stop program\r\n");
Console.Read();
service.OnStop();
return;
}
Run(service);
}
}
/// <summary>
/// On critical errors bail out
/// </summary>
/// <param name="errorMessage"></param>
/// <param name="exception"></param>
async Task OnCriticalError(ICriticalErrorContext context)
{
var fatalMessage = $"The following critical error was encountered:\n{context.Error}\nProcess is shutting down.";
try
{
Console.WriteLine("fatalMessage: " + fatalMessage);
TraceWriter.Error(fatalMessage);
if (context.Exception.InnerException != null)
{
TraceWriter.Error("innerException message: " + context.Exception.InnerException.Message + @"\n" +
context.Exception.InnerException.StackTrace);
}
await context.Stop().ConfigureAwait(false);
}
finally
{
Environment.FailFast(fatalMessage, context.Exception);
}
}
/// <summary>
/// Starting the service
/// </summary>
/// <param name="args"></param>
private async Task AsyncOnStart()
{
TraceWriter.Trace("AsyncOnStart() running .. ");
try
{
TraceWriter.Info("Running configuration management ..");
var config =
ConfigurationManager.OpenExeConfiguration(ConfigurationUserLevel.None);
var section =
config.GetSection("connectionStrings");
if (!section.SectionInformation.IsProtected)
{
section.SectionInformation.ProtectSection("DPAPIProtection");
section.SectionInformation.ForceSave = true;
config.Save(ConfigurationSaveMode.Modified);
}
var endpointConfig = new EndpointConfiguration("hyflo.service");
endpointConfig.EnableInstallers();
endpointConfig.SendFailedMessagesTo("hyflo.service.errors");
endpointConfig.AuditProcessedMessagesTo("hyflo.service.audit");
endpointConfig.UseSerialization<XmlSerializer>();
endpointConfig.DefineCriticalErrorAction(OnCriticalError);
endpointConfig.LimitMessageProcessingConcurrencyTo(8);
var persistenceMethod = ConfigurationManager.AppSettings["NServicebusPersistence"];
if (persistenceMethod.ToUpper() == "RAVENDB")
{
string connectionString = ConfigurationManager.ConnectionStrings["NServiceBus/Persistence"].ConnectionString;
TraceWriter.Info($"Setting persistence to RavenDBPersistence based on connectionString '{connectionString}' .. ");
var documentStore = new Raven.Client.Document.DocumentStore
{
ConnectionStringName = "NServiceBus/Persistence",
DefaultDatabase = "HyFlo.Service"
};
documentStore.Initialize();
endpointConfig.UsePersistence<RavenDBPersistence>().SetDefaultDocumentStore(documentStore);
}
else
{
TraceWriter.Info("Setting persistence to InMemoryPersistence .. ");
endpointConfig.UsePersistence<InMemoryPersistence>();
}
var transportConnection = ConfigurationManager.ConnectionStrings[@"NServiceBus/Transport"];
string transportConnectionString = transportConnection.ConnectionString;
if (String.IsNullOrEmpty(transportConnectionString))
{
transportConnectionString = "host=localhost";
}
TraceWriter.Info($"Configuring RabbitMQTransport for connection '{transportConnectionString}' .. ");
var transport = endpointConfig.UseTransport<RabbitMQTransport>();
transport.ConnectionString(transportConnectionString);
transport.UseConventionalRoutingTopology();
string databusBasePath = ConfigurationManager.AppSettings["DataBusBasePath"] ?? "";
TraceWriter.Info($"Setting Databus's basepath to '{databusBasePath}' ..");
endpointConfig.UseDataBus<FileShareDataBus>().BasePath(databusBasePath);
TraceWriter.Info("Scannning for Hyflo assemblies .. ");
List<string> hyfloAssemblies =
Directory.GetFiles(Directory.GetCurrentDirectory(), "HyFlo.*dll", SearchOption.TopDirectoryOnly).ToList();
TraceWriter.Info("Initializing Autofac with assemblies .. ");
foreach (string assemblyName in hyfloAssemblies)
{
TraceWriter.Info($"Scanning '{assemblyName}' for an Autofac module .. ");
}
try
{
var containerSettings = endpointConfig.UseContainer(new AutofacServiceProviderFactory());
containerSettings.ConfigureContainer(containerBuilder =>
{
var loadedAssemblies = hyfloAssemblies.Select(Assembly.LoadFile).ToList();
containerBuilder.RegisterAssemblyModules(loadedAssemblies.ToArray());
});
}
catch (Exception ex)
{
TraceWriter.Error($"{ex.Message}\n{ex.StackTrace}\n{ex.InnerException?.Source}");
await OnCriticalError(new CriticalErrorContext(null, $"Exception occurred during initialization. Exception is: {ex.Message}\n{ex.StackTrace}\n{ex.InnerException?.Source}", ex));
}
TraceWriter.Info("Setting up default message conventions ..");
var conventions = endpointConfig.Conventions();
conventions.DefiningTimeToBeReceivedAs(type => TimeSpan.FromMinutes(10));
var logFactory = LogManager.Use<DefaultFactory>();
logFactory.Level(LogLevel.Debug);
endpointConfig.EnableFeature<HyfloInitializationFeature>();
TraceWriter.Info("Initialized! Now starting Hyflo ..");
_endpointInstance = await Endpoint.Start(endpointConfig);
TraceWriter.Info("Hyflo started ..");
}
catch (Exception exception)
{
TraceWriter.Error($"{exception.Message}\n{exception.StackTrace}\n{exception.InnerException?.Source}");
await OnCriticalError(new CriticalErrorContext(null, "Failed to start the bus.", exception));
}
}
/// <summary>
/// Stopping the service
/// </summary>
/// <returns></returns>
private Task AsyncOnStop()
{
TraceWriter.Info("Shutting down Hyflo ..");
if (_endpointInstance != null)
{
return _endpointInstance.Stop();
}
return Task.CompletedTask;
}
// <summary>
/// Starting the service
/// </summary>
/// <param name="args"></param>
protected override void OnStart(string[] args)
{
TraceWriter.Trace("ProgramService.OnStart() ..");
AsyncOnStart().GetAwaiter().GetResult();
}
/// <summary>
/// Stopping the service
/// </summary>
protected override void OnStop()
{
TraceWriter.Trace("ProgramService.OnStop() ..");
AsyncOnStop().GetAwaiter().GetResult();
}
}
HyfloInitializationFeature 是一个启动任务，它将所有当前工作流的列表发送到连接的客户端。
using HyFlo.Tasks.Interfaces;
using NServiceBus.Features;
namespace HyFlo.Service
{
public class HyfloInitializationFeature : Feature
{
protected override void Setup(FeatureConfigurationContext context)
{
context.RegisterStartupTask(c => new SystemStartupTask(c.Build<IGetInfoOnAllWorkflowsTask>(), c.Build<ISystemTestTask>()));
}
}
}
这是系统启动任务：
namespace HyFlo.Service
{
public class SystemStartupTask : FeatureStartupTask, IDisposable
{
private readonly IGetInfoOnAllWorkflowsTask _getInfoOnAllWorkflowsTask;
private readonly ISystemTestTask _systemTestTask;
public SystemStartupTask(IGetInfoOnAllWorkflowsTask getInfoOnAllWorkflowsTask, ISystemTestTask systemTestTask)
{
_getInfoOnAllWorkflowsTask = getInfoOnAllWorkflowsTask;
_systemTestTask = systemTestTask;
}
public void Dispose()
{
}
/// <summary>
/// OnStart
/// </summary>
/// <param name="session"></param>
/// <returns></returns>
protected override async Task OnStart(IMessageSession session)
{
TraceWriter.Trace($"==> Running SystemStartupTask.OnStart() ..");
DeleteHeartbeatIndicator();
ObjectCache cache = MemoryCache.Default;
var policy = new CacheItemPolicy { SlidingExpiration = (TimeSpan.FromSeconds(15)) };
bool systemOk = false;
if (_systemTestTask != null)
{
_systemTestTask.ContextWrapper = new MessageContextWrapper { Session = session };
_systemTestTask.QuickTest = false;
_systemTestTask.TestComponentsMask = 31;
_systemTestTask.SystemStartup = true;
if (_systemTestTask.Run())
{
var systemResultTexts = new List<string>();
if (int.TryParse(_systemTestTask.Results.ToString(), out int systemsResult))
{
if ((systemsResult & 1) == 1)
{
systemResultTexts.Add("HSB offline");
}
if ((systemsResult & 2) == 2)
{
systemResultTexts.Add("HDM offline");
}
if ((systemsResult & 4) == 4)
{
systemResultTexts.Add("SqlServer offline");
}
if ((systemsResult & 8) == 8)
{
systemResultTexts.Add("Workflow Queueing offline");
}
}
TraceWriter.Trace(
$"SystemStartupTask: Results returned by systemtest: '{(!systemResultTexts.Any() ? "All online" : String.Join(",", systemResultTexts))}' ..");
if (!_systemTestTask.Failure && !systemResultTexts.Any())
{
TraceWriter.Info("HyFlo dependencies all up and running ..");
systemOk = true;
// for caching purposes
if (_getInfoOnAllWorkflowsTask != null)
{
_getInfoOnAllWorkflowsTask.UserId = "";
_getInfoOnAllWorkflowsTask.Run();
}
}
else
{
TraceWriter.Warn("HyFlo can't be started. One or more of its dependencies returned a failure!");
}
}
}
else
{
TraceWriter.Warn("A general failure occurred during the Hyflo systemcheck!");
}
var allworkflowStates = new WorkflowState[0];
if (_getInfoOnAllWorkflowsTask != null)
{
if (!_getInfoOnAllWorkflowsTask.Failure && _getInfoOnAllWorkflowsTask.Results is List<WorkflowState>)
{
allworkflowStates = (_getInfoOnAllWorkflowsTask.Results as List<WorkflowState>).ToArray();
}
TraceWriter.Info(
$"Workflowstate retrieval completed. Failure: '{_getInfoOnAllWorkflowsTask.Failure}', Results: '{allworkflowStates.Count()}' workflow{(allworkflowStates.Count() == 1 ? "" : "s")} found ..");
var timeStamp = DateTime.Now;
TraceWriter.Trace("Sending IStartedHyflo @ GUI with the following parameters:\n" +
$"\tErrorCode: '{Convert.ToInt32(_systemTestTask?.Results)}'\n" +
$"\tFailure: '{!systemOk}'\n" + $"\tTimeStamp: '{timeStamp}'\n" +
$"\tResulting workflowcount: '{allworkflowStates.Length}'\n");
await session.Publish<IStartedHyflo>(evt =>
{
evt.ErrorCode = Convert.ToInt32(_systemTestTask?.Results);
evt.Failure = !systemOk;
evt.TimeStamp = timeStamp;
evt.Result = new DataBusProperty<WorkflowState[]>(allworkflowStates);
}).ConfigureAwait(false);
cache.Set("SystemTest", systemOk, policy);
}
}
/// <summary>
/// OnStop
/// </summary>
/// <param name="session"></param>
/// <returns></returns>
protected override async Task OnStop(IMessageSession session)
{
TraceWriter.Trace($"==> Running SystemStartupTask.OnStop() ..");
TraceWriter.Info("Deleting heartbeat indicator .. ");
DeleteHeartbeatIndicator();
var timeStamp = DateTime.Now;
TraceWriter.Trace("Sending IShutdownHyflo @ GUI with the following parameters:\n" + "\tErrorCode: '0'\n" +
"\tFailure: 'false'\n" + $"\tTimeStamp: '{timeStamp}'");
await session.Publish<IShutdownHyflo>(evt =>
{
evt.ErrorCode = 0;
evt.Failure = false;
evt.TimeStamp = timeStamp;
}).ConfigureAwait(false);
TraceWriter.Info("HyFlo has shutted down .. ");
}
/// <summary>
/// Function deletes the heartbeat.txt file that indicates wether a heartbeat flow is running or not
/// </summary>
private void DeleteHeartbeatIndicator()
{
string stateFolder = ConfigurationManager.AppSettings["StateFolder"];
string fullStateFolder = stateFolder.IndexOf(":", StringComparison.Ordinal) == -1
? $"{AppDomain.CurrentDomain.BaseDirectory}{stateFolder}"
: stateFolder;
string fileName = $@"{fullStateFolder}\heartbeat.txt";
if (File.Exists(fileName))
{
File.Delete(fileName);
}
}
}
}
GetVersionHandler 如下：
namespace HyFlo.Service.CommandHandlers.Info
{
public class GetVersionHandler : HandlerBase, IHandleMessages<GetVersionCommand>
{
public GetVersionHandler(ISystemTestTask systemTestTask, ITaskBuilderController taskBuilderController)
{
_systemTestTask = systemTestTask;
_taskBuilderController = taskBuilderController;
}
public virtual async Task Handle(GetVersionCommand message, IMessageHandlerContext context)
{
if (message == null)
return;
await RunDirect(async () =>
{
var assemblyVersion = Assembly.GetAssembly(typeof(GetVersionHandler)).GetName().Version;
var reply = new GetVersionReply
{
Failure = false,
Version = $"{assemblyVersion.Major}.{assemblyVersion.Minor}",
TimeStamp = DateTime.Now,
TaskId = CommandEventId,
ErrorCode = 0
};
await Retry.ExecuteAsync(async () =>
{
var replyOptions = new ReplyOptions();
replyOptions.SetHeader("CorrelationId", CorrelationId);
await context.Reply(reply, replyOptions);
});
}, new MessageContextWrapper { HandlerContext = context });
}
}
}
RunDirect 方法位于 HandlerBase 类中，它所做的第一件事就是记录正在处理的新消息。
public async Task RunDirect(Action codeToRun, IMessageContextWrapper contextWrapper)
{
_taskBuilderController.HandlerBase = this;
CommandEventId = Guid.NewGuid();
ContextWrapper = contextWrapper;
CorrelationId = GetHeaderValue("CorrelationId");
string fullMethodName = GetType().ToString();
TraceWriter.Trace("= New Message =====================================================================================================================================");
TraceWriter.Trace($"CommandEventId:{CommandEventId}: New message handled by handler of type '{fullMethodName}' with CorrelationId '{CorrelationId}' .. ");
TraceWriter.Trace("===================================================================================================================================================");
...
最后是我用来发送这个 GetVersionCommand 的测试客户端：
namespace HPS_EndpointTest
{
internal class Program
{
static void Main(string[] args)
{
var endpointConfiguration = new EndpointConfiguration("Hyflo.Client.Test");
endpointConfiguration.EnableInstallers();
endpointConfiguration.UsePersistence<InMemoryPersistence>();
endpointConfiguration.PurgeOnStartup(true);
var transport = endpointConfiguration.UseTransport<RabbitMQTransport>();
endpointConfiguration.UseTransport<RabbitMQTransport>().ConnectionString("host=hswv0601;username=hyflo;password=hyflo");
endpointConfiguration.UseTransport<RabbitMQTransport>().UseDirectRoutingTopology();
IEndpointInstance endpointInstance = null;
AsyncPump.Run(async delegate
{
endpointInstance = await Endpoint.Start(endpointConfiguration).ConfigureAwait(false);
});
Console.WriteLine(DateTime.Now.ToString("dd-MM-yyyyTHH:mm:ss.ffff") + ": Client started ..");
var correlationId = Guid.NewGuid();
string destination = "hyflo.service";
var sendOptions = new SendOptions();
sendOptions.SetHeader("CorrelationId", correlationId.ToString());
sendOptions.SetDestination(destination);
sendOptions.RequireImmediateDispatch();
var versionMessage = new HyFlo.Service.Messages.Commands.Info.GetVersionCommand();
Console.WriteLine(DateTime.Now.ToString("dd-MM-yyyyTHH:mm:ss.ffff") + ": Sending message ..");
AsyncPump.Run(async delegate
{
if (endpointInstance != null)
{
await endpointInstance.Send(versionMessage, sendOptions).ConfigureAwait(false);
}
});
Console.WriteLine(DateTime.Now.ToString("dd-MM-yyyyTHH:mm:ss.ffff") + ": Message sent!");
while (true)
{
var key = Console.ReadKey();
if (key.Key == ConsoleKey.Enter)
{
break;
}
}
AsyncPump.Run(async delegate
{
await endpointInstance.Stop().ConfigureAwait(false);
});
}
}
public class RetrieveAllWorkflowsHandler : IHandleMessages<HyFlo.Service.Messages.Events.IRetrievedAllWorkflowsEvent>
{
public Task Handle(HyFlo.Service.Messages.Events.IRetrievedAllWorkflowsEvent message, IMessageHandlerContext context)
{
Console.WriteLine(DateTime.Now.ToString("dd-MM-yyyyTHH:mm:ss.ffff") + ": IRetrievedAllWorkflowEvents reply received ..");
return Task.CompletedTask;
}
}
public class RetrieveVersionHandler : IHandleMessages<GetVersionReply>
{
public Task Handle(GetVersionReply message, IMessageHandlerContext context)
{
Console.WriteLine(DateTime.Now.ToString("dd-MM-yyyyTHH:mm:ss.ffff") + ": GetVersionReply reply received ..");
return Task.CompletedTask;
}
}
}
我只是不明白为什么会出现这种延迟。
我现在正在做的是创建一个功能有限的全新端点（即仅支持 GetVersionCommand），看看会发生什么。如果没有延迟，我将扩展其他命令的支持，看看会发生什么。希望你们中的任何人看到这一点并拍拍他/她自己的头并回复“但你忘记了某某”。
提前致谢！！
【问题讨论】：
-
端点有 80 个 IHandleMessages 处理程序。我开始怀疑这是否是这里问题的一部分。我刚刚创建了一个只有 GetVersionHandler 处理程序的端点。这个端点快如闪电。我现在正在移植所有其他处理程序，看看是否会再次出现延迟。
-
可能是 80 个处理程序，但由于 NServiceBus 而不太可能。您的代码中发生的不仅仅是 NServiceBus。我强烈建议联系 support@particular.net 或在discuss.particular.net 上发布此问题，因为我们将能够在那里为您提供更好的支持。审计消息头中还有发送时间和接收时间以及完成时间处理。但很难说为什么会出现延误。
-
你是对的。这不是80个处理程序。我制作了一个轻量级端点，它支持所有消息和处理程序作为完整的端点，仅没有后端代码。这个轻量级端点也快如闪电。消息接收和处理到处理程序的时间仅为 0.1 秒或更短。
-
哦，是的，我在discuss.particular.net/t/… 上也有一个话题。您对审计队列的建议很有趣。我没有考虑过。我将检查创建了哪些审核消息。
标签： windows delay nservicebus