NserviceBus：事件处理程序被调用 15 次以上......仅当应用程序在容器中运行时

Question

我已经设置了一个事件（以及一个传奇中使用的几个命令）。我有 2 个 Web 应用程序和 1 个注册端点的控制台应用程序。所有这些都关闭了自动订阅。一个应用程序为该事件声明一个处理程序并手动订阅该事件。

当我在容器中运行应用程序并使用 sqlPersistence 和 RabbitMq 传输时，处理程序会被一遍又一遍地调用。当我在没有容器的情况下运行并且没有明确设置任何持久性并使用 LearningTransport 时，代码的行为符合预期。

我知道当异常发生时处理程序会被一遍又一遍地调用，但这取决于默认的重试次数（5？），并且处理程序中没有发生异常，所以我不认为这就是问题所在。

当从 Web 应用程序发布事件（向其处理程序发送命令时会出现同样的问题）时，处理程序会被一遍又一遍地调用相同的消息：消息 ID、相关 ID、对话 ID、发送时间、 ...一切都是一样的，我不知道为什么，也没有在互联网上找到其他点击。

我使用 autofac 作为容器，使用 NSB 7.0.1 和 RabbitMq 作为传输。所有应用程序和持久性（sql server）和传输都在 Docker 容器中运行。

对于每个应用程序，我都将其设置为相同，只是我只在一个中注册了处理程序，如下所示：

感谢您的帮助！

public virtual IServiceProvider ConfigureServices(IServiceCollection services)
        {
//...
var loggerFactory = 
            container.Resolve<Microsoft.Extensions.Logging.ILoggerFactory>();
            var logFactory = LogManager.Use<MicrosoftLogFactory>();
            logFactory.UseMsFactory(loggerFactory);

            var endpointConfiguration = new EndpointConfiguration(applicationName);
            endpointConfiguration.EnableInstallers();
            endpointConfiguration.SendFailedMessagesTo("error");
            endpointConfiguration.AuditProcessedMessagesTo("audit");
            endpointConfiguration.UseSerialization<NewtonsoftSerializer>();

            endpointConfiguration.DisableFeature<AutoSubscribe>();

            EnsureSqlDatabaseExists(persistenceConnectionString, container.Resolve<ILogger<IEndpointInstance>>());

            var persistence = endpointConfiguration.UsePersistence<SqlPersistence>();
            persistence.TablePrefix("Monrovo");
            var dialect = persistence.SqlDialect<SqlDialect.MsSqlServer>();
            dialect.Schema("receiver");

            persistence.ConnectionBuilder(() => new SqlConnection(persistenceConnectionString));

            var transport = endpointConfiguration.UseTransport<RabbitMQTransport>();
            transport.UseConventionalRoutingTopology();
            transport.ConnectionString(transportConnectionString);

            endpointConfiguration.UseContainer<AutofacBuilder>(customizations =>
            {
                customizations.ExistingLifetimeScope(container);
            });

            await ScriptRunner.Install(
                sqlDialect: new SqlDialect.MsSqlServer(),
                tablePrefix: "Monrovo",
                connectionBuilder: () => new SqlConnection(persistenceConnectionString),
                scriptDirectory: Path.GetDirectoryName(Assembly.GetEntryAssembly().Location) + @"/NServiceBus.Persistence.Sql/MsSqlServer",
                shouldInstallOutbox: true,
                shouldInstallSagas: true,
                shouldInstallSubscriptions: true,
                shouldInstallTimeouts: true);

            var startableEndpoint = await Endpoint.Create(endpointConfiguration);
            var endpoint = await startableEndpoint.Start();

_endpointInstance.Subscribe<ArticleApproved>().Wait();

            var updateBuilder = new ContainerBuilder();
            updateBuilder.RegisterInstance(_endpointInstance)
                .As<IMessageSession>().As<IEndpointInstance>().SingleInstance();

            // There is currently no workaround from NServiceBus for this.
            // See https://github.com/Particular/NServiceBus/issues/4421
#pragma warning disable CS0618 // Type or member is obsolete
            updateBuilder.Update(container);
#pragma warning restore CS0618 // Type or member is obsolete

            return new AutofacServiceProvider(container);
}


private static void EnsureSqlDatabaseExists(string connectionString, ILogger<IEndpointInstance> logger)
        {
            var builder = new SqlConnectionStringBuilder(connectionString);
            var originalCatalog = builder.InitialCatalog;

            builder.InitialCatalog = "master";
            var masterConnectionString = builder.ConnectionString;

            try
            {
                using (var connection = new SqlConnection(masterConnectionString))
                {
                    connection.Open();
                    var command = connection.CreateCommand();
                    command.CommandText =
                        $"IF NOT EXISTS (SELECT * FROM sys.databases WHERE name = '{originalCatalog}')" +
                        $"  CREATE DATABASE [{originalCatalog}]; ";
                    command.ExecuteNonQuery();
                }
            }
            catch (SqlException)
            {
                // Need to handle this better. Locally, we need to use master first. In Azure this is not possible or necessary
                // TODO: Build a sql docker container with a database already created.
                builder.InitialCatalog = originalCatalog;
                masterConnectionString = builder.ConnectionString;

                try
                {
                    using (var connection = new SqlConnection(masterConnectionString))
                    {
                        connection.Open();
                        var command = connection.CreateCommand();
                        command.CommandText =
                            $"IF NOT EXISTS (SELECT * FROM sys.databases WHERE name = '{originalCatalog}')" +
                            $"  CREATE DATABASE [{originalCatalog}]; ";
                        command.ExecuteNonQuery();
                    }
                }
                catch (SqlException innerException)
                {
                    logger.LogCritical(innerException, $"Unable to connect to SQL Server. Check that {builder.DataSource} is available");
                }
            }

            builder.InitialCatalog = originalCatalog;
            masterConnectionString = builder.ConnectionString;

            try
            {
                using (var connection = new SqlConnection(masterConnectionString))
                {
                    connection.Open();

                    var command = connection.CreateCommand();
                    command.CommandText =
                        $"IF NOT EXISTS(SELECT 1 FROM sys.schemas WHERE name = 'receiver')" +
                        $"BEGIN EXEC sys.sp_executesql N'CREATE SCHEMA receiver;' END";
                    command.ExecuteNonQuery();
                }
            }
            catch { }
        }



public class test : IHandleMessages<ArticleApproved>
    {
        public Task Handle(RfpApproved message, IMessageHandlerContext context)
        {
            // Initially handles the published event and starts the saga
            context.SendLocal(new StartRFPSync(message.RfpId));
            return Task.CompletedTask;
        }
    }

Answer 1

问题最终是在断点处停止时抛出异常（我认为是超时）。 NSB 心跳设置为 5 秒，因此延长开发时间可以让您解决这个问题，但我已经确认当我让代码运行时我的问题不存在。通常调试代码是一件好事，但异步进程会导致