将 Mass Transit 与 Azure 服务总线主题一起使用时，消费者未收到消息

Question

我刚开始学习Masstransit，我想问你一个查询on the accepted answer in this post。在这里，我有同样的问题，消费者没有收到消息。 可能是我犯了一些错误或误解了一些基本概念。

我做了什么：

• 在 Azure 服务总线中创建主题

• 为主题创建订阅

• 根据建议here向主题发送消息。我可以从 Azure 门户看到消息已成功添加到主题中。

• 现在我正在尝试使用以下代码来使用该消息：

      private static async Task SetupMassTransit()
    {
        var services = new ServiceCollection();
        services.AddMassTransit(x =>
        {
            x.AddConsumer<MessageConsumer>(); //Consumer class
            x.UsingAzureServiceBus((context, cfg) =>
            {
                cfg.Host(ServiceBusConnectionString);
                // https://stackoverflow.com/questions/64229653/receiver-not-picking-up-message-with-mass-transit-subscription
                cfg.SubscriptionEndpoint(TopicName, SubscriptionName, 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<MessageConsumer>(context);
                });
            });
        });
            var provider = services.BuildServiceProvider();
            IBusControl busControl = provider.GetRequiredService<IBusControl>();
            busControl.Start();
    }
  

• 我希望在运行消费者项目时应该调用“Consume”函数。但事实并非如此。虽然我无法通过故障消费者阅读它，但它也会创建故障主题。

    public class MessageConsumer : IConsumer<SendMessage>
    { 
    public Task Consume(ConsumeContext<SendMessage> context)
    {
        Console.WriteLine($"Message processed: SequenceNumber:{context.Message.customerId} - Text:{context.Message.messages}");
        return Task.CompletedTask;
    }
    }
  

请注意：消费者和生产者使用相同的消息对象（即在同一个命名空间内）

你能告诉我我到底在哪里犯了错误吗？ 提前谢谢

Answer 1

我认为您在为服务总线配置之前添加了消费者（需要仔细检查）。我做了一个不同的设置，我使用了 RecieveEndpoint 而不是 SubscriptionEndpoint，它可以工作。

                services.AddMassTransit(x =>
                {
                    x.UsingAzureServiceBus((context, cfg) =>
                    {
                        cfg.Host(ServiceBusConnectionString);
                        cfg.ReceiveEndpoint(typeof(SendMessage).Name.ToLower(), endpoint =>
                        {
                            endpoint.ConfigureConsumer<MessageConsumer >(context);
                        });
                    });
                    x.AddConsumer<MessageConsumer >();
                });
                services.AddMassTransitHostedService();

其他一切都作为你的例子。

（编辑）：我确实意识到一个用于主题，另一个用于队列，所以请注意这一点。

Answer 2

谢谢丹尼尔。 DI 可能是问题所在，尽管我注意到它现在在没有 DI 的情况下工作（我将尝试找出它第一次无法工作的原因） 一件事，正如提到here

cfg.SubscriptionEndpoint("your-topic-name", "your-subscription-name", e =>

我觉得应该是的

cfg.SubscriptionEndpoint("your-subscription-name", "your-topic-name", e =>

最终的工作源代码：

        private static async Task SetupMassTransit(IServiceCollection services)
    {
        try
        {
            //services.AddScoped<IConsumer<SendMessage>, MessageConsumer>(); It's still working without this line
            services.AddMassTransit(x =>
            {
                x.AddConsumer<MessageConsumer>();
                x.UsingAzureServiceBus((context, cfg) =>
                {
                    cfg.Host(ServiceBusConnectionString);
                    cfg.SubscriptionEndpoint(SubscriptionName, TopicName, e =>
                    {
                       e.ConfigureConsumer<MessageConsumer>(context);
                    });
                });
            });
            var provider = services.BuildServiceProvider();
            IBusControl busControl = provider.GetRequiredService<IBusControl>();
            busControl.Start();
        }
        catch (Exception e)
        {
            Console.WriteLine(e);
            throw;
        }
    }