azure - 接收者未通过公共(public)交通接收消息 - 订阅

Question

我正在使用下面的代码来设置 MassTransit 以使用 ServiceBus

private static ServiceProvider SetupServiceCollection()
{
    var connectionString = ConfigurationManager.AppSettings["AzureServiceBusConnectionString"];
    var services = new ServiceCollection()
        .AddMassTransit(x =>
        {
            x.UsingAzureServiceBus((context, cfg) =>
            {
                cfg.Host(connectionString);
                cfg.ConfigureEndpoints(context);
                cfg.Message<MyMessage>(x =>
                {
                    x.SetEntityName("my-topic");
                });
            });
        });

    return services.BuildServiceProvider(); 
}

我使用以下代码发送消息

var message = new MyMessage()
{
    MessageIdentifier = Guid.NewGuid().ToString(),
};

await _busControl.Publish(message);

我希望我的消息仅发送到我的主题

但是，MassTransit 正在创建主题，名称似乎是使用类型名称生成的。我怎样才能完全阻止这种情况？

我正在按如下方式设置接收器

public static void SetupMassTransit(this ServiceCollection services, string connectionString)
{
    services.AddMassTransit(x =>
    {
        x.UsingAzureServiceBus((context, cfg) =>
        {
            cfg.Host(connectionString);
            cfg.ConfigureEndpoints(context);
            x.UsingAzureServiceBus((context, cfg) =>
            {
                cfg.Host(connectionString);
                cfg.SubscriptionEndpoint<MyMessage>("low", e =>
                {
                    e.Consumer<MyMessageConsumer>(context);
                    e.PrefetchCount = 100;
                    e.MaxConcurrentCalls = 100;
                    e.LockDuration = TimeSpan.FromMinutes(5);
                    e.MaxAutoRenewDuration = TimeSpan.FromMinutes(30);
                    e.UseMessageRetry(r => r.Intervals(100, 200, 500, 800, 1000));
                    e.UseInMemoryOutbox();
                    e.ConfigureConsumeTopology = false;
                });
        });
    });
}

我可以看到消息正在正确发送，如 Service Bus Explorer 中的订阅内所示。然而，接收器没有接收到它？没有错误或有什么要继续的吗？真是令人沮丧

保罗

Answer 1

您正在调用ConfigureEndpoints，默认情况下它将为已添加的消费者、saga 等创建接收端点。但是，您的代码示例未显示任何 .AddConsumer 方法。如果您没有任何使用者，请不要调用 ConfigureEndpoints。

对于您的接收器，您应该使用:

public static void SetupMassTransit(this ServiceCollection services, string connectionString)
{
    services.AddMassTransit(x =>
    {
        x.AddConsumer<MyMessageConsumer>();
        
        x.UsingAzureServiceBus((context, cfg) =>
        {
            cfg.Host(connectionString);

            cfg.SubscriptionEndpoint("your-topic-name", "your-subscription-name", e =>
            {
                e.PrefetchCount = 100;
                e.MaxConcurrentCalls = 100;
                e.LockDuration = TimeSpan.FromMinutes(5);
                e.MaxAutoRenewDuration = TimeSpan.FromMinutes(30);

                e.UseMessageRetry(r => r.Intervals(100, 200, 500, 800, 1000));
                e.UseInMemoryOutbox();

                e.ConfigureConsumer<MyMessageConsumer>(context);
            });
        });
    });
}

对于您的制作人，您可以简单地使用:

private static ServiceProvider SetupServiceCollection()
{
    var connectionString = ConfigurationManager.AppSettings["AzureServiceBusConnectionString"];
    var services = new ServiceCollection()
        .AddMassTransit(x =>
        {
            x.UsingAzureServiceBus((context, cfg) =>
            {
                cfg.Host(connectionString);
            });
        });

    return services.BuildServiceProvider(); 
}

然后，使用上面创建的 IServiceProvider 进行发布:

var bus = serviceProvider.GetRequiredService<IBus>();
var endpoint = await bus.GetSendEndpoint(new Uri("topic:your-topic-name"));
await endpoint.Send(new MyMessage());

这应该满足您所需的绝对最低要求。