接收方未通过大众运输接收消息-订阅 [英] Receiver not picking up message with Mass Transit - Subscription

问题描述

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

I am using the code below to setup MassTransit to use 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(); 
}

我使用以下代码发送消息

I use the following code to send a message

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

await _busControl.Publish(message);

我希望将邮件仅发送到我的主题

I want my message to be sent to my-topic only

但是，MassTransit正在创建主题，名称似乎是使用类型名称生成的.我该如何完全停止呢?

However, MassTransit is creating topic, the names seem to be getting generated using the type names. How do I totally stop this?

我正在如下设置接收器

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订阅中所示.但是，接收器没有接听吗?没有错误或其他任何事情吗?真令人沮丧

I can see that the message is being sent correctly, as its shown inside the subscription in Service Bus Explorer. However, the receiver is not picking it up? There are no errors or anything to go on? Really frustrating

保罗

推荐答案

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

You're calling ConfigureEndpoints, which will by default create receive endpoints for the consumers, sagas, etc. that have been added. However, your code sample doesn't show any .AddConsumer methods. If you don't have any consumers, don't call ConfigureEndpoints.

对于接收者，您应该使用:

For your receiver, you should use:

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);
            });
        });
    });
}

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

For your producer, you can simply use:

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 进行发布:

Then, to publish, using your IServiceProvider that was created above:

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

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

That should do the absolute minimum required that you need.

这篇关于接收方未通过大众运输接收消息-订阅的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！