在Azure Function中配置MassTransit 8.x接收处理消息的问题

我有一个Azure Function，一直以来都在接收并处理另一应用发送的常规格式消息。接触到MassTransit后，希望在ASP.NET应用和这个Azure Function接收端都用上它，但没法在Handler里正确配置消息解码和处理逻辑。之前试过相关方案，但在MassTransit 8.x版本里不生效。

我们团队要求必须保留全Azure Function架构，虽然知道用带后台服务的App Service监听队列是最佳方案，但还是需要Azure Function下的正确配置方法。

Handler #

public class MessageReceiverHandler
{
    private readonly ILogger<MessageReceiverHandler> _logger;
    private readonly IKlaviyoNameFormSender _klaviyoNameFormSender;
    private readonly IConsumer<MyOrderItemFormFilledPayload> _consumer;
    public MessageReceiverHandler(ILogger<MessageReceiverHandler> logger, IKlaviyoNameFormSender klaviyoNameFormSender, IConsumer<MyOrderItemFormFilledPayload> consumer)
    {
        _logger = logger;
        _klaviyoNameFormSender = klaviyoNameFormSender;
        _consumer = consumer;
    }

    [Function(nameof(MessageReceiverHandler))]
    public async Task Run(
        [ServiceBusTrigger("sbq-klaviyo-orderitem-filled-form", Connection = "ServiceBusConnection")]
        ServiceBusReceivedMessage message,
        ServiceBusMessageActions messageActions)
    {
        _logger.LogInformation("Message ID: {id}", message.MessageId);
        _logger.LogInformation("Message Body: {body}", message.Body);

      
        var orderItem = JsonConvert.DeserializeObject<MessageBody>(message.Body.ToString());
        await _klaviyoNameFormSender.SendKlaviyoNameFormSender(orderItem.OrderItemId, default);

        await messageActions.CompleteMessageAsync(message);
    }
}

Registration #

public static IHostBuilder RegisterServices(this IHostBuilder builder)
{
    builder.ConfigureServices((context, services) =>
    {
        services.AddScoped<IKlaviyoNameFormSender, KlaviyoNameFormSender>();
        services.AddScoped<ITTGLinq2DbRepository, TTGLinq2DbRepository>();

        services.Configure<AzureServiceBusSettings>(context.Configuration.GetSection("AzureServiceBusSettings"));
        services.AddApplicationInsightsTelemetryWorkerService();
        services.ConfigureFunctionsApplicationInsights();
        services.Configure<KlaviyoOption>(opt =>
        {
[...omiss...]
        });

        services.AddMassTransit(x =>
        {
            x.AddConsumer<KlaviyoNameCollectionMessageConsumer>();
            x.UsingAzureServiceBus((ctx, cfg) =>
                {
                    var settings = ctx.GetRequiredService<IOptions<AzureServiceBusSettings>>().Value;

                    cfg.Host(settings.ConnectionString);

                    cfg.ReceiveEndpoint(settings.QueueName, e =>
                    {
                        e.ConfigureConsumer<KlaviyoNameCollectionMessageConsumer>(ctx);
                    });
                });
        });
    });
    return builder;
}

Consumer #

public class KlaviyoNameCollectionMessageConsumer
    : IConsumer<KlaviyoOrderItemFormFilledPayload>
{
    public async Task Consume(ConsumeContext<KlaviyoOrderItemFormFilledPayload> context)
    {
        return;
        //TO IMPLEMENT
    }
}

public class KlaviyoOrderItemFormFilledPayload
{
    public int  OrderItemId { get; set; }
}

内容的提问来源于stack exchange，提问作者advapi