将 NServiceBus 与 SqlTransport 一起使用 - 无法验证存储在 SQL 中的消息答案

【问题标题】：Using NServiceBus with SqlTransport - Can't verify message stored in SQL将 NServiceBus 与 SqlTransport 一起使用 - 无法验证存储在 SQL 中的消息
【发布时间】：2020-04-12 01:16:14
【问题描述】：

我正在使用 SQLTransport 浏览 NServiceBus 上的示例，我显然没有正确理解传输的概念，因为我没有看到可以在此处找到的示例应用程序之后的预期行为：https://docs.particular.net/samples/sqltransport-sqlpersistence/

我无法查看示例数据消息OrderSubmitted 的存储位置。我的假设是，由于我使用的是 SqlTransport，我应该能够看到存储在表 (Samples.Sql.Sender) 中的消息，该表表示 OrderSubmitted 项目的队列。示例中有一个发送者和接收者，并且我禁用了接收者，以便我可以验证消息是否存储在 SQL Server 上的表（队列）中。当发件人启动时，我看到一个sender.Samples.Sql.Sender 表出现。我假设这是此端点发送的消息的队列。但是，当您发送消息时，我在 sender.Samples.Sql.Sender 表（或与此相关的任何表）中看不到记录，但它表明 OrderSubmitted 已发布，那么它去了哪里？

我基本上有一个生产者/消费者模式，我想用 NServiceBus 来实现。我有生产者会在持久队列上生成消息。然后我有一个或多个工作服务器，它们会将项目从队列中取出并处理它们。因为我有多个工作服务器都在查看同一个队列来执行工作，所以我需要确保其中只有一个可以处理任何给定的消息（我不需要主题，我只想要一个持久队列的简单实现，其中许多工作进程可以帮助处理队列中的项目。NServiceBus 可以用于这种类型的用例吗。似乎可以，但我在如何配置这种类型的环境方面有点挣扎。应该如何配置它被视为工作队列，而不是每个实例都尝试处理相同消息的主题？

using System;
using System.Threading.Tasks;
using NServiceBus;
using NServiceBus.Transport.SQLServer;

public static class Program
{
    static Random random;

    public static async Task Main()
    {
        random = new Random();

        Console.Title = "Samples.Sql.Sender";
        var endpointConfiguration = new EndpointConfiguration("Samples.Sql.Sender");
        endpointConfiguration.SendFailedMessagesTo("error");
        endpointConfiguration.EnableInstallers();

        #region SenderConfiguration

        var connection = @"Data Source=MySqlServer;Initial Catalog=NServiceBus;Integrated Security=True;Max Pool Size=100";
        var transport = endpointConfiguration.UseTransport<SqlServerTransport>();
        transport.ConnectionString(connection);
        transport.DefaultSchema("sender");
        transport.UseSchemaForQueue("error", "dbo");
        transport.UseSchemaForQueue("audit", "dbo");
        transport.NativeDelayedDelivery().DisableTimeoutManagerCompatibility();

        endpointConfiguration.UsePersistence<InMemoryPersistence>();

        var subscriptions = transport.SubscriptionSettings();
        subscriptions.SubscriptionTableName(
            tableName: "Subscriptions", 
            schemaName: "dbo");

        #endregion

        SqlHelper.CreateSchema(connection, "sender");

        var endpointInstance = await Endpoint.Start(endpointConfiguration)
            .ConfigureAwait(false);
        Console.WriteLine("Press enter to send a message");
        Console.WriteLine("Press any key to exit");

        while (true)
        {
            var key = Console.ReadKey();
            Console.WriteLine();

            if (key.Key != ConsoleKey.Enter)
            {
                break;
            }

            var orderSubmitted = new OrderSubmitted
            {
                OrderId = Guid.NewGuid(),
                Value = random.Next(100)
            };
            await endpointInstance.Publish(orderSubmitted)
                .ConfigureAwait(false);
            Console.WriteLine("Published OrderSubmitted message");
        }
        await endpointInstance.Stop()
            .ConfigureAwait(false);
    }
}

【问题讨论】：

标签： nservicebus asp.net-core-2.1

【解决方案1】：

看看这里，也许它有你正在寻找的细节？ https://docs.particular.net/transports/sql/#how-it-works

一些概念：

端点有一个队列当您在代理传输（如 SQL、Azure、Rabbit MQ）中发送消息时，一条消息将保存在目标端点队列名称中一旦目标端点运行，它将从该队列中读取并删除消息...

很高兴在电话中提供更多帮助，请直接给我发电子邮件 sean.farmar@particular.net

【讨论】：